parallel.c

Go to the documentation of this file.

/*
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */
 
/**
 * $Id: f3ec4650a1b416fee4686f8e66c35f41b26b1727 $
 *
 * @file unlang/parallel.c
 * @brief Implementation of the unlang "parallel" keyword.
 *
 * @copyright 2006-2019 The FreeRADIUS server project
 */
RCSID("$Id: f3ec4650a1b416fee4686f8e66c35f41b26b1727 $")
 
#include "function.h"
#include "interpret_priv.h"
#include "module_priv.h"
#include "parallel_priv.h"
#include "subrequest_priv.h"
 
 
/** Cancel a specific child
 *
 */
static inline CC_HINT(always_inline) void unlang_parallel_cancel_child(unlang_parallel_state_t *state, int i)
{
        request_t *child = state->children[i].request;
        request_t *request = child->parent;     /* For debug messages */
 
        switch (state->children[i].state) {
        case CHILD_INIT:
                state->children[i].state = CHILD_CANCELLED;
                fr_assert(!state->children[i].request);
                break;
 
        case CHILD_EXITED:
                state->children[i].state = CHILD_CANCELLED;
                TALLOC_FREE(state->children[i].request);
                break;
 
        case CHILD_RUNNABLE:    /* Don't check runnable_id, may be yielded */
                /*
                 *      Signal the child to stop
                 *
                 *      The signal function cleans up the request
                 *      and signals anything that was tracking it
                 *      that it's now complete.
                 */
                unlang_interpret_signal(child, FR_SIGNAL_CANCEL);
 
                /*
                 *      Free it.
                 */
                TALLOC_FREE(state->children[i].request);
                break;
 
        case CHILD_DONE:
                fr_assert(!fr_heap_entry_inserted(child->runnable_id));
 
                /*
                 *      Completed children just get freed
                 */
                TALLOC_FREE(state->children[i].request);
                break;
 
        case CHILD_DETACHED:
        case CHILD_CANCELLED:
                return;
        }
 
        RDEBUG3("parallel - child %s (%d/%d) CANCELLED",
                state->children[i].name,
                i + 1, state->num_children);
        state->children[i].state = CHILD_CANCELLED;
}
 
#if 0
/** Cancel all the child's siblings
 *
 * Siblings will be excluded from final result calculation for the parallel section.
 */
static void unlang_parallel_cancel_siblings(request_t *request)
{
        unlang_stack_t          *stack = request->parent->stack;
        unlang_stack_frame_t    *frame = &stack->frame[stack->depth];
        unlang_parallel_state_t *state = talloc_get_type_abort(frame->state, unlang_parallel_state_t);
        int i;
 
        for (i = 0; i < state->num_children; i++) {
                if (state->children[i].request == request) continue;    /* Don't cancel this one */
 
                unlang_parallel_cancel_child(state, i);
        }
}
#endif
 
/** Signal handler to deal with UNLANG_ACTION_DETACH
 *
 * When a request detaches we need
 */
static void unlang_parallel_child_signal(request_t *request, UNUSED fr_signal_t action, void *uctx)
{
        unlang_parallel_child_t         *child = uctx;
        unlang_stack_frame_t            *frame;
        unlang_parallel_state_t         *state;
 
        frame = frame_current(request->parent);
        state = talloc_get_type_abort(frame->state, unlang_parallel_state_t);
 
        RDEBUG3("parallel - child %s (%d/%d) DETACHED",
                request->name,
                child->num + 1, state->num_children);
 
        child->state = CHILD_DETACHED;
        child->request = NULL;
        state->num_complete++;
 
        /*
         *      All children exited, resume the parent
         */
        if (state->num_complete == state->num_children) {
                RDEBUG3("Signalling parent %s that all children have EXITED or DETACHED", request->parent->name);
                unlang_interpret_mark_runnable(request->parent);
        }
 
        return;
}
 
/** When the chld is done, tell the parent that we've exited.
 *
 */
static unlang_action_t unlang_parallel_child_done(UNUSED rlm_rcode_t *p_result, UNUSED int *p_priority, request_t *request, void *uctx)
{
        unlang_parallel_child_t *child = uctx;
 
        /*
         *      If we have a parent, then we're running synchronously
         *      with it.  Tell the parent that we've exited, and it
         *      can continue.
         *
         *      Otherwise we're a detached child, and we don't tell
         *      the parent anything.  Because we have that kind of
         *      relationship.
         */
        if (child->state == CHILD_RUNNABLE) {
                /*
                 *      Reach into the parent to get the unlang_parallel_state_t
                 *      for the whole parallel block.
                 */
                unlang_stack_frame_t            *frame = frame_current(request->parent);
                unlang_parallel_state_t         *state = talloc_get_type_abort(frame->state, unlang_parallel_state_t);
 
                RDEBUG3("parallel - child %s (%d/%d) EXITED",
                        request->name,
                        child->num + 1, state->num_children);
 
                child->state = CHILD_EXITED;
                state->num_complete++;
 
                /*
                 *      All children exited, resume the parent
                 */
                if (state->num_complete == state->num_children) {
                        RDEBUG3("Signalling parent %s that all children have EXITED or DETACHED", request->parent->name);
                        unlang_interpret_mark_runnable(request->parent);
                }
        }
 
        /*
         *      Don't change frame->result, it's the result of the child.
         */
        return UNLANG_ACTION_CALCULATE_RESULT;
}
 
static unlang_action_t unlang_parallel_resume(rlm_rcode_t *p_result, request_t *request, unlang_stack_frame_t *frame)
{
        unlang_parallel_state_t         *state = talloc_get_type_abort(frame->state, unlang_parallel_state_t);
 
        int                             i, priority;
        rlm_rcode_t                     result;
 
        for (i = 0; i < state->num_children; i++) {
                if (!state->children[i].request) continue;
 
                fr_assert(state->children[i].state == CHILD_EXITED);
                REQUEST_VERIFY(state->children[i].request);
 
                RDEBUG3("parallel - child %s (%d/%d) DONE",
                        state->children[i].name,
                        i + 1, state->num_children);
 
                state->children[i].state = CHILD_DONE;
 
                priority = ((unlang_stack_t *)state->children[i].request->stack)->priority;
                result = ((unlang_stack_t *)state->children[i].request->stack)->result;
 
                /*
                 *      Return isn't allowed to make it back
                 *      to the parent... Not sure this is
                 *      the correct behaviour, but it's what
                 *      was there before.
                 */
                if (priority == MOD_ACTION_RETURN) {
                        priority = 0;
                } else if (priority == MOD_ACTION_REJECT) {
                        result = RLM_MODULE_REJECT;
                        priority = 0;
                }
 
                /*
                 *      Do priority over-ride.
                 */
                if (priority > state->priority) {
                        state->result = result;
                        state->priority = priority;
 
                        RDEBUG4("** [%i] %s - over-riding result from higher priority to (%s %d)",
                                stack_depth_current(request), __FUNCTION__,
                                fr_table_str_by_value(mod_rcode_table, result, "<invalid>"),
                                priority);
                }
        }
 
        /*
         *      Reap the children....
         */
        for (i = 0; i < state->num_children; i++) {
                if (!state->children[i].request) continue;
 
                fr_assert(!fr_heap_entry_inserted(state->children[i].request->runnable_id));
                TALLOC_FREE(state->children[i].request);
        }
 
        *p_result = state->result;
        return UNLANG_ACTION_CALCULATE_RESULT;
}
 
/** Run one or more sub-sections from the parallel section.
 *
 */
static unlang_action_t unlang_parallel_process(rlm_rcode_t *p_result, request_t *request, unlang_stack_frame_t *frame)
{
        unlang_parallel_state_t         *state = talloc_get_type_abort(frame->state, unlang_parallel_state_t);
        request_t                       *child;
        int                             i;
 
        /*
         *      If the children should be created detached, we return
         *      "noop".  This function then creates the children,
         *      detaches them, and returns.
         */
        if (state->detach) {
                state->priority = 0;
                state->result = RLM_MODULE_NOOP;
        }
 
        /*
         *      Loop over all the children.
         *
         *      We always service the parallel section from top to
         *      bottom, and we always service all of it.
         */
        for (i = 0; i < state->num_children; i++) {
                fr_assert(state->children[i].instruction != NULL);
                child = unlang_io_subrequest_alloc(request,
                                                   request->dict, state->detach);
                child->packet->code = request->packet->code;
 
                RDEBUG3("parallel - child %s (%d/%d) INIT",
                        child->name,
                        i + 1, state->num_children);
 
                if (state->clone) {
                        /*
                         *      Note that we do NOT copy the
                         *      Session-State list!  That
                         *      contains state information for
                         *      the parent.
                         */
                        if ((fr_pair_list_copy(child->request_ctx,
                                               &child->request_pairs,
                                               &request->request_pairs) < 0) ||
                            (fr_pair_list_copy(child->reply_ctx,
                                               &child->reply_pairs,
                                               &request->reply_pairs) < 0) ||
                            (fr_pair_list_copy(child->control_ctx,
                                               &child->control_pairs,
                                               &request->control_pairs) < 0)) {
                                REDEBUG("failed copying lists to clone");
                        error:
                                /*
                                 *      Detached children which have
                                 *      already been created are
                                 *      allowed to continue.
                                 */
                                if (!state->detach) {
                                        /*
                                         *      Remove the current child
                                         *
                                         *      Should also free detached children
                                         */
                                        unlang_interpret_request_done(child);
 
                                        /*
                                         *      Remove all previously
                                         *      spawned children.
                                         */
                                        for (--i; i >= 0; i--) unlang_interpret_request_done(child);
                                }
 
                                RETURN_MODULE_FAIL;
                        }
                }
 
                /*
                 *      Child starts detached, the parent knows
                 *      and can exit immediately once all
                 *      the children are initialised.
                 */
                if (state->detach) {
                        if (RDEBUG_ENABLED3) {
                                request_t *parent = request;
 
                                request = child;
                                RDEBUG3("parallel - child %s (%d/%d) DETACHED",
                                        request->name,
                                        i + 1, state->num_children);
                                request = parent;
                        }
 
                        state->children[i].state = CHILD_DETACHED;
 
                        /*
                         *      Detach the child, and insert
                         *      it into the backlog.
                         */
                        if ((unlang_subrequest_lifetime_set(child) < 0) || (request_detach(child) < 0)) {
                                request = child;
                                RPEDEBUG("Failed detaching request");
                                talloc_free(child);
 
                                RETURN_MODULE_FAIL;
                        }
                /*
                 *      If the children don't start detached
                 *      push a function onto the stack to
                 *      notify the parent when the child is
                 *      done.
                 */
                } else {
                        unlang_stack_frame_t *child_frame;
 
                        if (unlang_function_push(child,
                                                 NULL,
                                                 unlang_parallel_child_done,
                                                 unlang_parallel_child_signal,
                                                 ~FR_SIGNAL_DETACH,
                                                 UNLANG_TOP_FRAME,
                                                 &state->children[i]) < 0) goto error;
                        child_frame = frame_current(child);
                        return_point_set(child_frame);          /* Don't unwind this frame */
 
                        state->children[i].num = i;
                        state->children[i].name = talloc_bstrdup(state, child->name);
                        state->children[i].request = child;
                        state->children[i].state = CHILD_RUNNABLE;
 
                }
                interpret_child_init(child);
 
                /*
                 *      Push the first instruction for
                 *      the child to run.
                 */
                if (unlang_interpret_push(child,
                                          state->children[i].instruction, RLM_MODULE_FAIL,
                                          UNLANG_NEXT_STOP,
                                          state->detach ? UNLANG_TOP_FRAME : UNLANG_SUB_FRAME) < 0) goto error;
        }
 
        /*
         *      If all children start detached,
         *      then we're done.
         */
        if (state->detach) return UNLANG_ACTION_CALCULATE_RESULT;
 
        /*
         *      Don't call this function again when
         *      the parent resumes, instead call
         *      a function to process the results
         *      of the children.
         */
        frame_repeat(frame, unlang_parallel_resume);
 
        /*
         *      Yield to the children
         *
         *      They scamper off to play on their
         *      own when they're all done, the last
         *      one tells the parent, so it can resume,
         *      and gather up all the results.
         */
        return UNLANG_ACTION_YIELD;
}
 
/** Send a signal from parent request to all of it's children
 *
 */
static void unlang_parallel_signal(UNUSED request_t *request,
                                   unlang_stack_frame_t *frame, fr_signal_t action)
{
        unlang_parallel_state_t *state = talloc_get_type_abort(frame->state, unlang_parallel_state_t);
        int                     i;
 
        if (action == FR_SIGNAL_CANCEL) {
                for (i = 0; i < state->num_children; i++) unlang_parallel_cancel_child(state, i);
 
                return;
        }
 
        /*
         *      Signal all of the runnable/running children.
         */
        for (i = 0; i < state->num_children; i++) {
                if (state->children[i].state != CHILD_RUNNABLE) continue;
 
                unlang_interpret_signal(state->children[i].request, action);
        }
}
 
static unlang_action_t unlang_parallel(rlm_rcode_t *p_result, request_t *request, unlang_stack_frame_t *frame)
{
        unlang_t const                  *instruction;
        unlang_group_t                  *g;
        unlang_parallel_t               *gext;
        unlang_parallel_state_t         *state;
 
        int                             i;
 
        g = unlang_generic_to_group(frame->instruction);
        if (!g->num_children) {
                *p_result = RLM_MODULE_NOOP;
                return UNLANG_ACTION_CALCULATE_RESULT;
        }
 
        gext = unlang_group_to_parallel(g);
 
        /*
         *      Allocate an array for the children.
         */
        MEM(frame->state = state = _talloc_zero_pooled_object(request,
                                                              sizeof(unlang_parallel_state_t) +
                                                              (sizeof(state->children[0]) * g->num_children),
                                                              "unlang_parallel_state_t",
                                                              g->num_children,
                                                              (talloc_array_length(request->name) * 2)));
        if (!state) {
                *p_result = RLM_MODULE_FAIL;
                return UNLANG_ACTION_CALCULATE_RESULT;
        }
 
        (void) talloc_set_type(state, unlang_parallel_state_t);
        state->result = RLM_MODULE_FAIL;
        state->priority = -1;                           /* as-yet unset */
        state->detach = gext->detach;
        state->clone = gext->clone;
        state->num_children = g->num_children;
 
        /*
         *      Initialize all of the children.
         */
        for (i = 0, instruction = g->children; instruction != NULL; i++, instruction = instruction->next) {
                state->children[i].state = CHILD_INIT;
                state->children[i].instruction = instruction;
        }
 
        frame->process = unlang_parallel_process;
        return unlang_parallel_process(p_result, request, frame);
}
 
void unlang_parallel_init(void)
{
        unlang_register(UNLANG_TYPE_PARALLEL,
                           &(unlang_op_t){
                                .name = "parallel",
                                .interpret = unlang_parallel,
                                .signal = unlang_parallel_signal,
                                .rcode_set = true,
                                .debug_braces = true
                           });
}