From: Alan T. DeKok Date: Tue, 10 Jan 2017 15:11:25 +0000 (-0500) Subject: fr_worker_decode_request() is no longer a separate function X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=8aac00e9359ca8fedeaa6e265d05723e1d7910ca;p=thirdparty%2Ffreeradius-server.git fr_worker_decode_request() is no longer a separate function it's only called from one place --- diff --git a/src/util/worker.c b/src/util/worker.c index d5222c212a3..990d8b482ff 100644 --- a/src/util/worker.c +++ b/src/util/worker.c @@ -377,131 +377,6 @@ static void fr_worker_nak(fr_worker_t *worker, fr_channel_data_t *cd, fr_time_t } -/** Decode a request from either the localized queue, or the to_decode queue - * - * The request returned from this function MUST be immediately runnable. - * - * @param[in] worker the worker - * @param[in] now the current time - * @return - * - NULL on nothing to decode - * - REQUEST the decoded request - */ -static REQUEST *fr_worker_decode_request(fr_worker_t *worker, fr_time_t now) -{ - int rcode; - fr_channel_data_t *cd; - REQUEST *request; -#ifndef HAVE_TALLOC_POOLED_OBJECT - TALLOC_CTX *ctx; -#endif - - /* - * Find either a localized message, or one which is in - * the "to_decode" queue. - */ -redo: - WORKER_HEAP_POP(localized, cd, request.list); - if (!cd) { - WORKER_HEAP_POP(to_decode, cd, request.list); - } - if (!cd) return NULL; - - worker->num_decoded++; - - /* - * This message has asynchronously aged out while it was - * in the queue. Delete it, and go get another one. - */ - if (cd->request.start_time && (cd->m.when != *cd->request.start_time)) { - MPRINT("\tIGNORING old message\n"); - fr_worker_nak(worker, cd, fr_time()); - goto redo; - } - -#ifndef HAVE_TALLOC_POOLED_OBJECT - /* - * Get a talloc pool specifically for this packet. - */ - ctx = talloc_pool(worker, worker->talloc_pool_size); - if (!ctx) goto nak; - - talloc_set_name_const(ctx, "REQUEST"); - - request = (REQUEST *) ctx; - memset(request, 0, sizeof(*request)); -#else - request = ctx = talloc_pooled_object(worker, REQUEST, 1, worker->talloc_pool_size); - if (!request) goto nak; -#endif - - /* - * Receive a message to the worker queue, and decode it - * to a request. - */ - rad_assert(cd->transport <= worker->num_transports); - rad_assert(worker->transports[cd->transport] != NULL); - - /* - * Update the transport-specific fields. - * - * Note that the message "when" time MUST be copied from - * the original recv time. We use "when" here, instead - * of *cd->request.recv_time, on the odd chance that a - * new packet arrived while we were getting around to - * processing this message. - */ - request->channel = cd->channel.ch; - request->transport = worker->transports[cd->transport]; - request->original_recv_time = cd->request.start_time; - request->recv_time = cd->m.when; - request->priority = cd->priority; - request->runnable = worker->runnable; - request->el = worker->el; - request->packet_ctx = cd->ctx; - - /* - * Now that the "request" structure has been initialized, go decode the packet. - */ - rcode = worker->transports[cd->transport]->decode(cd->ctx, cd->m.data, cd->m.data_size, request); - if (rcode < 0) { - MPRINT("\tFAILED decode of request %zd\n", request->number); - talloc_free(ctx); -nak: - fr_worker_nak(worker, cd, fr_time()); - return NULL; - } - - /* - * Hoist run-time checks here. - */ - if (!cd->request.start_time) request->original_recv_time = &request->recv_time; - - /* - * We're done with this message. - */ - fr_message_done(&cd->m); - - /* - * New requests are inserted into the time order list in - * strict time priority. Once they are in the list, they - * are only removed when the request is freed. - */ - FR_DLIST_INSERT_HEAD(worker->time_order, request->time_order); - - /* - * Bootstrap the async state machine with the initial - * state of the request. The process_async function will - * take care of pushing the state machine through it's - * transitions. - */ - request->process_async = request->transport->process; - fr_time_tracking_start(&request->tracking, now); - - return request; -} - - /** Reply to a request * * And clean it up. @@ -726,7 +601,12 @@ static void fr_worker_check_timeouts(fr_worker_t *worker, fr_time_t now) */ static REQUEST *fr_worker_get_request(fr_worker_t *worker, fr_time_t now) { + int rcode; + fr_channel_data_t *cd; REQUEST *request; +#ifndef HAVE_TALLOC_POOLED_OBJECT + TALLOC_CTX *ctx; +#endif /* * Grab a runnable request, and resume it. @@ -738,9 +618,108 @@ static REQUEST *fr_worker_get_request(fr_worker_t *worker, fr_time_t now) } /* - * Grab a request to decode, and start it. + * Find either a localized message, or one which is in + * the "to_decode" queue. */ - return fr_worker_decode_request(worker, now); +redo: + WORKER_HEAP_POP(localized, cd, request.list); + if (!cd) { + WORKER_HEAP_POP(to_decode, cd, request.list); + } + if (!cd) return NULL; + + worker->num_decoded++; + + /* + * This message has asynchronously aged out while it was + * in the queue. Delete it, and go get another one. + */ + if (cd->request.start_time && (cd->m.when != *cd->request.start_time)) { + MPRINT("\tIGNORING old message\n"); + fr_worker_nak(worker, cd, fr_time()); + goto redo; + } + +#ifndef HAVE_TALLOC_POOLED_OBJECT + /* + * Get a talloc pool specifically for this packet. + */ + ctx = talloc_pool(worker, worker->talloc_pool_size); + if (!ctx) goto nak; + + talloc_set_name_const(ctx, "REQUEST"); + + request = (REQUEST *) ctx; + memset(request, 0, sizeof(*request)); +#else + request = ctx = talloc_pooled_object(worker, REQUEST, 1, worker->talloc_pool_size); + if (!request) goto nak; +#endif + + /* + * Receive a message to the worker queue, and decode it + * to a request. + */ + rad_assert(cd->transport <= worker->num_transports); + rad_assert(worker->transports[cd->transport] != NULL); + + /* + * Update the transport-specific fields. + * + * Note that the message "when" time MUST be copied from + * the original recv time. We use "when" here, instead + * of *cd->request.recv_time, on the odd chance that a + * new packet arrived while we were getting around to + * processing this message. + */ + request->channel = cd->channel.ch; + request->transport = worker->transports[cd->transport]; + request->original_recv_time = cd->request.start_time; + request->recv_time = cd->m.when; + request->priority = cd->priority; + request->runnable = worker->runnable; + request->el = worker->el; + request->packet_ctx = cd->ctx; + + /* + * Now that the "request" structure has been initialized, go decode the packet. + */ + rcode = worker->transports[cd->transport]->decode(cd->ctx, cd->m.data, cd->m.data_size, request); + if (rcode < 0) { + MPRINT("\tFAILED decode of request %zd\n", request->number); + talloc_free(ctx); +nak: + fr_worker_nak(worker, cd, fr_time()); + return NULL; + } + + /* + * Hoist run-time checks here. + */ + if (!cd->request.start_time) request->original_recv_time = &request->recv_time; + + /* + * We're done with this message. + */ + fr_message_done(&cd->m); + + /* + * New requests are inserted into the time order list in + * strict time priority. Once they are in the list, they + * are only removed when the request is freed. + */ + FR_DLIST_INSERT_HEAD(worker->time_order, request->time_order); + + /* + * Bootstrap the async state machine with the initial + * state of the request. The process_async function will + * take care of pushing the state machine through it's + * transitions. + */ + request->process_async = request->transport->process; + fr_time_tracking_start(&request->tracking, now); + + return request; }