Line data    Source code

       1             : /*
       2             :    Unix SMB/CIFS implementation.
       3             :    Infrastructure for async requests
       4             :    Copyright (C) Volker Lendecke 2008
       5             :    Copyright (C) Stefan Metzmacher 2009
       6             : 
       7             :      ** NOTE! The following LGPL license applies to the tevent
       8             :      ** library. This does NOT imply that all of Samba is released
       9             :      ** under the LGPL
      10             : 
      11             :    This library is free software; you can redistribute it and/or
      12             :    modify it under the terms of the GNU Lesser General Public
      13             :    License as published by the Free Software Foundation; either
      14             :    version 3 of the License, or (at your option) any later version.
      15             : 
      16             :    This library is distributed in the hope that it will be useful,
      17             :    but WITHOUT ANY WARRANTY; without even the implied warranty of
      18             :    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
      19             :    Lesser General Public License for more details.
      20             : 
      21             :    You should have received a copy of the GNU Lesser General Public
      22             :    License along with this library; if not, see <http://www.gnu.org/licenses/>.
      23             : */
      24             : 
      25             : #include "replace.h"
      26             : #include "tevent.h"
      27             : #include "tevent_internal.h"
      28             : #include "tevent_util.h"
      29             : 
      30             : #undef tevent_queue_add
      31             : #undef tevent_queue_add_entry
      32             : #undef tevent_queue_add_optimize_empty
      33             : 
      34             : struct tevent_queue_entry {
      35             :         struct tevent_queue_entry *prev, *next;
      36             :         struct tevent_queue *queue;
      37             : 
      38             :         bool triggered;
      39             : 
      40             :         struct tevent_req *req;
      41             :         struct tevent_context *ev;
      42             : 
      43             :         tevent_queue_trigger_fn_t trigger;
      44             :         const char *trigger_name;
      45             :         void *private_data;
      46             :         uint64_t tag;
      47             : };
      48             : 
      49             : struct tevent_queue {
      50             :         const char *name;
      51             :         const char *location;
      52             : 
      53             :         bool running;
      54             :         struct tevent_immediate *immediate;
      55             : 
      56             :         size_t length;
      57             :         struct tevent_queue_entry *list;
      58             : };
      59             : 
      60             : static void tevent_queue_immediate_trigger(struct tevent_context *ev,
      61             :                                            struct tevent_immediate *im,
      62             :                                            void *private_data);
      63             : 
      64    13174990 : static int tevent_queue_entry_destructor(struct tevent_queue_entry *e)
      65             : {
      66    13174990 :         struct tevent_queue *q = e->queue;
      67             : 
      68    13174990 :         if (!q) {
      69           0 :                 return 0;
      70             :         }
      71             : 
      72    13174990 :         tevent_trace_queue_callback(q->list->ev, e, TEVENT_EVENT_TRACE_DETACH);
      73    13306440 :         tevent_thread_call_depth_notify(TEVENT_CALL_FLOW_REQ_QUEUE_LEAVE,
      74    13043540 :                                         q->list->req,
      75    13174990 :                                         q->list->req->internal.call_depth,
      76             :                                         e->trigger_name);
      77    13174990 :         DLIST_REMOVE(q->list, e);
      78    13174990 :         q->length--;
      79             : 
      80    13174990 :         if (!q->running) {
      81        2622 :                 return 0;
      82             :         }
      83             : 
      84    13172341 :         if (!q->list) {
      85    11527467 :                 return 0;
      86             :         }
      87             : 
      88     1590192 :         if (q->list->triggered) {
      89         453 :                 return 0;
      90             :         }
      91             : 
      92     1587687 :         tevent_schedule_immediate(q->immediate,
      93             :                                   q->list->ev,
      94             :                                   tevent_queue_immediate_trigger,
      95       74689 :                                   q);
      96             : 
      97     1587687 :         return 0;
      98             : }
      99             : 
     100     2309296 : static int tevent_queue_destructor(struct tevent_queue *q)
     101             : {
     102     2309296 :         q->running = false;
     103             : 
     104     2309298 :         while (q->list) {
     105           2 :                 struct tevent_queue_entry *e = q->list;
     106           2 :                 talloc_free(e);
     107             :         }
     108             : 
     109     2309296 :         return 0;
     110             : }
     111             : 
     112     2316920 : struct tevent_queue *_tevent_queue_create(TALLOC_CTX *mem_ctx,
     113             :                                           const char *name,
     114             :                                           const char *location)
     115             : {
     116       15477 :         struct tevent_queue *queue;
     117             : 
     118     2316920 :         queue = talloc_zero(mem_ctx, struct tevent_queue);
     119     2316920 :         if (!queue) {
     120           0 :                 return NULL;
     121             :         }
     122             : 
     123     2316920 :         queue->name = talloc_strdup(queue, name);
     124     2316920 :         if (!queue->name) {
     125           0 :                 talloc_free(queue);
     126           0 :                 return NULL;
     127             :         }
     128     2316920 :         queue->immediate = tevent_create_immediate(queue);
     129     2316920 :         if (!queue->immediate) {
     130           0 :                 talloc_free(queue);
     131           0 :                 return NULL;
     132             :         }
     133             : 
     134     2316920 :         queue->location = location;
     135             : 
     136             :         /* queue is running by default */
     137     2316920 :         queue->running = true;
     138             : 
     139     2316920 :         talloc_set_destructor(queue, tevent_queue_destructor);
     140     2316920 :         return queue;
     141             : }
     142             : 
     143     2552091 : static void tevent_queue_immediate_trigger(struct tevent_context *ev,
     144             :                                            struct tevent_immediate *im,
     145             :                                            void *private_data)
     146             : {
     147       82412 :         struct tevent_queue *q =
     148     2552091 :                 talloc_get_type_abort(private_data,
     149             :                 struct tevent_queue);
     150             : 
     151     2552091 :         if (!q->running) {
     152           0 :                 return;
     153             :         }
     154             : 
     155     2552091 :         if (!q->list) {
     156           0 :                 return;
     157             :         }
     158             : 
     159     2552091 :         tevent_trace_queue_callback(ev, q->list,
     160             :                                     TEVENT_EVENT_TRACE_BEFORE_HANDLER);
     161             :         /* Set the call depth of the request coming from the queue. */
     162     2634503 :         tevent_thread_call_depth_notify(TEVENT_CALL_FLOW_REQ_QUEUE_TRIGGER,
     163     2469679 :                                         q->list->req,
     164     2552091 :                                         q->list->req->internal.call_depth,
     165     2552091 :                                         q->list->trigger_name);
     166     2552091 :         q->list->triggered = true;
     167     2552091 :         q->list->trigger(q->list->req, q->list->private_data);
     168             : }
     169             : 
     170           1 : static void tevent_queue_noop_trigger(struct tevent_req *req,
     171             :                                       void *_private_data)
     172             : {
     173             :         /* this is doing nothing but blocking the queue */
     174           1 : }
     175             : 
     176    13174993 : static struct tevent_queue_entry *tevent_queue_add_internal(
     177             :                                         struct tevent_queue *queue,
     178             :                                         struct tevent_context *ev,
     179             :                                         struct tevent_req *req,
     180             :                                         tevent_queue_trigger_fn_t trigger,
     181             :                                         const char *trigger_name,
     182             :                                         void *private_data,
     183             :                                         bool allow_direct)
     184             : {
     185      131450 :         struct tevent_queue_entry *e;
     186             : 
     187    13174993 :         e = talloc_zero(req, struct tevent_queue_entry);
     188    13174993 :         if (e == NULL) {
     189           0 :                 return NULL;
     190             :         }
     191             : 
     192             :         /*
     193             :          * if there is no trigger, it is just a blocker
     194             :          */
     195    13174993 :         if (trigger == NULL) {
     196           1 :                 trigger = tevent_queue_noop_trigger;
     197             :         }
     198             : 
     199    13174993 :         e->queue = queue;
     200    13174993 :         e->req = req;
     201    13174993 :         e->ev = ev;
     202    13174993 :         e->trigger = trigger;
     203    13174993 :         e->trigger_name = trigger_name;
     204    13174993 :         e->private_data = private_data;
     205             : 
     206    13174993 :         if (queue->length > 0) {
     207             :                 /*
     208             :                  * if there are already entries in the
     209             :                  * queue do not optimize.
     210             :                  */
     211     1590208 :                 allow_direct = false;
     212             :         }
     213             : 
     214    13174993 :         if (req->async.fn != NULL) {
     215             :                 /*
     216             :                  * If the caller wants to optimize for the
     217             :                  * empty queue case, call the trigger only
     218             :                  * if there is no callback defined for the
     219             :                  * request yet.
     220             :                  */
     221           0 :                 allow_direct = false;
     222             :         }
     223             : 
     224    13174993 :         DLIST_ADD_END(queue->list, e);
     225    13174993 :         queue->length++;
     226    13174993 :         talloc_set_destructor(e, tevent_queue_entry_destructor);
     227    13174993 :         tevent_trace_queue_callback(ev, e, TEVENT_EVENT_TRACE_ATTACH);
     228    13174993 :         tevent_thread_call_depth_notify(TEVENT_CALL_FLOW_REQ_QUEUE_ENTER,
     229             :                                         req,
     230             :                                         req->internal.call_depth,
     231             :                                         e->trigger_name);
     232             : 
     233    13174993 :         if (!queue->running) {
     234          24 :                 return e;
     235             :         }
     236             : 
     237    13174963 :         if (queue->list->triggered) {
     238     1271909 :                 return e;
     239             :         }
     240             : 
     241             :         /*
     242             :          * If allowed we directly call the trigger
     243             :          * avoiding possible delays caused by
     244             :          * an immediate event.
     245             :          */
     246    11838141 :         if (allow_direct) {
     247    10620272 :                 tevent_trace_queue_callback(ev,
     248             :                                             queue->list,
     249             :                                             TEVENT_EVENT_TRACE_BEFORE_HANDLER);
     250    10620272 :                 queue->list->triggered = true;
     251    10620272 :                 queue->list->trigger(queue->list->req,
     252    10573296 :                                      queue->list->private_data);
     253    10620272 :                 return e;
     254             :         }
     255             : 
     256     1217869 :         tevent_schedule_immediate(queue->immediate,
     257             :                                   queue->list->ev,
     258             :                                   tevent_queue_immediate_trigger,
     259       19555 :                                   queue);
     260             : 
     261     1217869 :         return e;
     262             : }
     263             : 
     264           0 : bool tevent_queue_add(struct tevent_queue *queue,
     265             :                       struct tevent_context *ev,
     266             :                       struct tevent_req *req,
     267             :                       tevent_queue_trigger_fn_t trigger,
     268             :                       void *private_data)
     269             : {
     270           0 :         return _tevent_queue_add(queue, ev, req, trigger, NULL, private_data);
     271             : }
     272             : 
     273     1031117 : bool _tevent_queue_add(struct tevent_queue *queue,
     274             :                       struct tevent_context *ev,
     275             :                       struct tevent_req *req,
     276             :                       tevent_queue_trigger_fn_t trigger,
     277             :                       const char* trigger_name,
     278             :                       void *private_data)
     279             : {
     280        7750 :         struct tevent_queue_entry *e;
     281             : 
     282     1031117 :         e = tevent_queue_add_internal(queue, ev, req,
     283             :                                       trigger, trigger_name,
     284             :                                       private_data, false);
     285     1031117 :         if (e == NULL) {
     286           0 :                 return false;
     287             :         }
     288             : 
     289     1023367 :         return true;
     290             : }
     291             : 
     292           0 : struct tevent_queue_entry *tevent_queue_add_entry(
     293             :                                         struct tevent_queue *queue,
     294             :                                         struct tevent_context *ev,
     295             :                                         struct tevent_req *req,
     296             :                                         tevent_queue_trigger_fn_t trigger,
     297             :                                         void *private_data)
     298             : {
     299           0 :         return _tevent_queue_add_entry(queue, ev, req,
     300             :                                        trigger, NULL,
     301             :                                        private_data);
     302             : }
     303             : 
     304       94364 : struct tevent_queue_entry *_tevent_queue_add_entry(
     305             :                                         struct tevent_queue *queue,
     306             :                                         struct tevent_context *ev,
     307             :                                         struct tevent_req *req,
     308             :                                         tevent_queue_trigger_fn_t trigger,
     309             :                                         const char* trigger_name,
     310             :                                         void *private_data)
     311             : {
     312       94364 :         return tevent_queue_add_internal(queue, ev, req,
     313             :                                          trigger, trigger_name,
     314             :                                          private_data, false);
     315             : }
     316             : 
     317           0 : struct tevent_queue_entry *tevent_queue_add_optimize_empty(
     318             :                                         struct tevent_queue *queue,
     319             :                                         struct tevent_context *ev,
     320             :                                         struct tevent_req *req,
     321             :                                         tevent_queue_trigger_fn_t trigger,
     322             :                                         void *private_data)
     323             : {
     324           0 :         return _tevent_queue_add_optimize_empty(queue, ev, req,
     325             :                                                 trigger, NULL,
     326             :                                                 private_data);
     327             : }
     328             : 
     329    12049512 : struct tevent_queue_entry *_tevent_queue_add_optimize_empty(
     330             :                                         struct tevent_queue *queue,
     331             :                                         struct tevent_context *ev,
     332             :                                         struct tevent_req *req,
     333             :                                         tevent_queue_trigger_fn_t trigger,
     334             :                                         const char* trigger_name,
     335             :                                         void *private_data)
     336             : {
     337    12049512 :         return tevent_queue_add_internal(queue, ev, req,
     338             :                                          trigger, trigger_name,
     339             :                                          private_data, true);
     340             : }
     341             : 
     342           3 : void tevent_queue_entry_untrigger(struct tevent_queue_entry *entry)
     343             : {
     344           3 :         if (entry->queue->running) {
     345           0 :                 abort();
     346             :         }
     347             : 
     348           3 :         if (entry->queue->list != entry) {
     349           0 :                 abort();
     350             :         }
     351             : 
     352           3 :         entry->triggered = false;
     353           3 : }
     354             : 
     355       68160 : void tevent_queue_start(struct tevent_queue *queue)
     356             : {
     357       68160 :         if (queue->running) {
     358             :                 /* already started */
     359       68157 :                 return;
     360             :         }
     361             : 
     362           3 :         queue->running = true;
     363             : 
     364           3 :         if (!queue->list) {
     365           0 :                 return;
     366             :         }
     367             : 
     368           3 :         if (queue->list->triggered) {
     369           0 :                 return;
     370             :         }
     371             : 
     372           3 :         tevent_schedule_immediate(queue->immediate,
     373             :                                   queue->list->ev,
     374             :                                   tevent_queue_immediate_trigger,
     375           0 :                                   queue);
     376             : }
     377             : 
     378      117335 : void tevent_queue_stop(struct tevent_queue *queue)
     379             : {
     380      117335 :         queue->running = false;
     381      117335 : }
     382             : 
     383    73989734 : size_t tevent_queue_length(struct tevent_queue *queue)
     384             : {
     385    73989734 :         return queue->length;
     386             : }
     387             : 
     388           0 : bool tevent_queue_running(struct tevent_queue *queue)
     389             : {
     390           0 :         return queue->running;
     391             : }
     392             : 
     393             : struct tevent_queue_wait_state {
     394             :         uint8_t dummy;
     395             : };
     396             : 
     397             : static void tevent_queue_wait_trigger(struct tevent_req *req,
     398             :                                       void *private_data);
     399             : 
     400      409750 : struct tevent_req *tevent_queue_wait_send(TALLOC_CTX *mem_ctx,
     401             :                                           struct tevent_context *ev,
     402             :                                           struct tevent_queue *queue)
     403             : {
     404        7041 :         struct tevent_req *req;
     405        7041 :         struct tevent_queue_wait_state *state;
     406        7041 :         bool ok;
     407             : 
     408      409750 :         req = tevent_req_create(mem_ctx, &state,
     409             :                                 struct tevent_queue_wait_state);
     410      409750 :         if (req == NULL) {
     411           0 :                 return NULL;
     412             :         }
     413             : 
     414      409750 :         ok = _tevent_queue_add(queue, ev, req,
     415             :                               tevent_queue_wait_trigger,
     416             :                               "tevent_queue_wait_trigger",
     417             :                               NULL);
     418      409750 :         if (!ok) {
     419           0 :                 tevent_req_oom(req);
     420           0 :                 return tevent_req_post(req, ev);
     421             :         }
     422             : 
     423      402709 :         return req;
     424             : }
     425             : 
     426      409639 : static void tevent_queue_wait_trigger(struct tevent_req *req,
     427             :                                          void *private_data)
     428             : {
     429      409639 :         tevent_req_done(req);
     430      409639 : }
     431             : 
     432      409537 : bool tevent_queue_wait_recv(struct tevent_req *req)
     433             : {
     434        7034 :         enum tevent_req_state state;
     435        7034 :         uint64_t err;
     436             : 
     437      409537 :         if (tevent_req_is_error(req, &state, &err)) {
     438           0 :                 tevent_req_received(req);
     439           0 :                 return false;
     440             :         }
     441             : 
     442      409537 :         tevent_req_received(req);
     443      409537 :         return true;
     444             : }
     445             : 
     446      391843 : void tevent_queue_entry_set_tag(struct tevent_queue_entry *qe, uint64_t tag)
     447             : {
     448      391843 :         if (qe == NULL) {
     449           0 :                 return;
     450             :         }
     451             : 
     452      391843 :         qe->tag = tag;
     453             : }
     454             : 
     455      391847 : uint64_t tevent_queue_entry_get_tag(const struct tevent_queue_entry *qe)
     456             : {
     457      391847 :         if (qe == NULL) {
     458           1 :                 return 0;
     459             :         }
     460             : 
     461      391846 :         return qe->tag;
     462             : }