usr/actor.c

   1 /*
   2  * iSCSI usermode single-threaded scheduler
   3  *
   4  * Copyright (C) 2004 Dmitry Yusupov, Alex Aizman
   5  * maintained by open-iscsi@googlegroups.com
   6  *
   7  * This program is free software; you can redistribute it and/or modify
   8  * it under the terms of the GNU General Public License as published
   9  * by the Free Software Foundation; either version 2 of the License, or
  10  * (at your option) any later version.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  15  * General Public License for more details.
  16  *
  17  * See the file COPYING included with this distribution for more details.
  18  */
  19 #include <inttypes.h>
  20 #include "actor.h"
  21 #include "log.h"
  22 #include "list.h"
  23
  24 static LIST_HEAD(pend_list);
  25 static LIST_HEAD(poll_list);
  26 static LIST_HEAD(actor_list);
  27 static volatile uint64_t previous_time;
  28 static volatile uint32_t scheduler_loops;
  29 static volatile int poll_in_progress;
  30 static volatile uint64_t actor_jiffies = 0;
  31
  32 #define actor_diff(_time1, _time2) ({ \
  33         uint64_t __ret; \
  34         if ((_time2) >= (_time1)) \
  35            __ret = (_time2) - (_time1); \
  36         else \
  37            __ret = ((~0ULL) - (_time1)) + (_time2); \
  38         __ret; \
  39 })
  40
  41 #define ACTOR_TICKS             actor_jiffies
  42 #define ACTOR_TICKS_10MS(_a)    (_a)
  43 #define ACTOR_MS_TO_TICKS(_a)   ((_a)/ACTOR_RESOLUTION)
  44
  45 static uint64_t
  46 actor_diff_time(actor_t *thread, uint64_t current_time)
  47 {
  48         uint64_t diff_time = actor_diff(thread->scheduled_at, current_time);
  49         if(diff_time >= thread->ttschedule)
  50                 return 0;
  51         return (thread->ttschedule - diff_time);
  52 }
  53
  54 #define time_after(a,b) \
  55         ((int64_t)(b) - (int64_t)(a) < 0)
  56
  57 void
  58 actor_init(void)
  59 {
  60         poll_in_progress = 0;
  61         previous_time = 0;
  62         scheduler_loops = 0;
  63 }
  64
  65 void
  66 actor_new(actor_t *thread, void (*callback)(void *), void *data)
  67 {
  68         INIT_LIST_HEAD(&thread->list);
  69         thread->state = ACTOR_NOTSCHEDULED;
  70         thread->callback = callback;
  71         thread->data = data;
  72 }
  73
  74 void
  75 actor_delete(actor_t *thread)
  76 {
  77         log_debug(7, "thread %08lx delete: state %d", (long)thread,
  78                         thread->state);
  79         switch(thread->state) {
  80         case ACTOR_SCHEDULED:
  81         case ACTOR_WAITING:
  82         case ACTOR_POLL_WAITING:
  83                 log_debug(1, "deleting a scheduled/waiting thread!");
  84                 list_del_init(&thread->list);
  85                 break;
  86         default:
  87                 break;
  88         }
  89         thread->state = ACTOR_NOTSCHEDULED;
  90 }
  91
  92 static void
  93 actor_schedule_private(actor_t *thread, uint32_t ttschedule, int head)
  94 {
  95         uint64_t delay_time, current_time;
  96         actor_t *next_thread;
  97
  98         delay_time = ACTOR_MS_TO_TICKS(ttschedule);
  99         current_time = ACTOR_TICKS;
 100
 101         log_debug(7, "thread %p schedule: delay %" PRIu64 " state %d",
 102                 thread, delay_time, thread->state);
 103
 104         /* convert ttscheduled msecs in 10s of msecs by dividing for now.
 105          * later we will change param to 10s of msecs */
 106         switch(thread->state) {
 107         case ACTOR_WAITING:
 108                 log_error("rescheduling a waiting thread!");
 109                 list_del(&thread->list);
 110         case ACTOR_NOTSCHEDULED:
 111                 INIT_LIST_HEAD(&thread->list);
 112                 /* if ttschedule is 0, put in scheduled queue and change
 113                  * state to scheduled, else add current time to ttschedule and
 114                  * insert in the queue at the correct point */
 115                 if (delay_time == 0) {
 116                         /* For head addition, it must go onto the head of the
 117                            actor_list regardless if poll is in progress or not
 118                          */
 119                         if (poll_in_progress && !head) {
 120                                 thread->state = ACTOR_POLL_WAITING;
 121                                 list_add_tail(&thread->list,
 122                                               &poll_list);
 123                         } else {
 124                                 thread->state = ACTOR_SCHEDULED;
 125                                 if (head)
 126                                         list_add(&thread->list,
 127                                                  &actor_list);
 128                                 else
 129                                         list_add_tail(&thread->list,
 130                                                       &actor_list);
 131                         }
 132                 } else {
 133                         thread->state = ACTOR_WAITING;
 134                         thread->ttschedule = delay_time;
 135                         thread->scheduled_at = current_time;
 136
 137                         /* insert new entry in sort order */
 138                         list_for_each_entry(next_thread, &pend_list, list) {
 139                                 log_debug(7, "thread %p %" PRIu64 " %"PRIu64,
 140                                         next_thread,
 141                                         next_thread->scheduled_at +
 142                                         next_thread->ttschedule,
 143                                         current_time + delay_time);
 144
 145                                 if (time_after(next_thread->scheduled_at +
 146                                                        next_thread->ttschedule,
 147                                                 current_time + delay_time)) {
 148                                         list_add(&thread->list,
 149                                                  &next_thread->list);
 150                                         goto done;
 151                                 }
 152                         }
 153
 154                         list_add_tail(&thread->list, &pend_list);
 155                 }
 156 done:
 157                 break;
 158         case ACTOR_POLL_WAITING:
 159         case ACTOR_SCHEDULED:
 160                 // don't do anything
 161                 break;
 162         case ACTOR_INVALID:
 163                 log_error("BUG: Trying to schedule a thread that has not been "
 164                           "setup. Ignoring sched.");
 165                 break;
 166         }
 167
 168 }
 169
 170 void
 171 actor_schedule_head(actor_t *thread)
 172 {
 173         actor_schedule_private(thread, 0, 1);
 174 }
 175
 176 void
 177 actor_schedule(actor_t *thread)
 178 {
 179         actor_schedule_private(thread, 0, 0);
 180 }
 181
 182 void
 183 actor_timer(actor_t *thread, uint32_t timeout, void (*callback)(void *),
 184             void *data)
 185 {
 186         actor_new(thread, callback, data);
 187         actor_schedule_private(thread, timeout, 0);
 188 }
 189
 190 int
 191 actor_timer_mod(actor_t *thread, uint32_t timeout, void *data)
 192 {
 193         if (thread->state == ACTOR_WAITING) {
 194                 list_del_init(&thread->list);
 195                 thread->data = data;
 196                 actor_schedule_private(thread, timeout, 0);
 197                 return 1;
 198         }
 199         return 0;
 200 }
 201
 202 void
 203 actor_check(uint64_t current_time)
 204 {
 205         struct actor *thread, *tmp;
 206
 207         list_for_each_entry_safe(thread, tmp, &pend_list, list) {
 208                 if (actor_diff_time(thread, current_time)) {
 209                         log_debug(7, "thread %08lx wait some more",
 210                                 (long)thread);
 211                         /* wait some more */
 212                         break;
 213                 }
 214
 215                 /* it is time to schedule this entry */
 216                 list_del_init(&thread->list);
 217
 218                 log_debug(2, "thread %08lx was scheduled at %" PRIu64 ":"
 219                         "%" PRIu64 ", curtime %" PRIu64 " q_forw %p "
 220                         "&pend_list %p",
 221                         (long)thread, thread->scheduled_at, thread->ttschedule,
 222                         current_time, pend_list.next, &pend_list);
 223
 224                 if (poll_in_progress) {
 225                         thread->state = ACTOR_POLL_WAITING;
 226                         list_add_tail(&thread->list, &poll_list);
 227                         log_debug(7, "thread %08lx now in poll_list",
 228                                 (long)thread);
 229                 } else {
 230                         thread->state = ACTOR_SCHEDULED;
 231                         list_add_tail(&thread->list, &actor_list);
 232                         log_debug(7, "thread %08lx now in actor_list",
 233                                 (long)thread);
 234                 }
 235         }
 236 }
 237
 238 void
 239 actor_poll(void)
 240 {
 241         uint64_t current_time;
 242         struct actor *thread;
 243
 244         /* check that there are no any concurrency */
 245         if (poll_in_progress) {
 246                 log_error("concurrent actor_poll() is not allowed");
 247         }
 248
 249         /* don't check wait list every single poll.
 250          * get new time. Shift it to make 10s of msecs approx
 251          * if new time is not same as old time */
 252         if (scheduler_loops++ > ACTOR_MAX_LOOPS) {
 253                 /* try coming in about every 100 msecs */
 254                 current_time = ACTOR_TICKS;
 255                 scheduler_loops = 0;
 256                 /* checking whether we are in the same tick... */
 257                 if ( ACTOR_TICKS_10MS(current_time) !=
 258                      ACTOR_TICKS_10MS(previous_time)) {
 259                         previous_time = current_time;
 260                         actor_check(current_time);
 261                 }
 262         }
 263
 264         /* the following code to check in the main data path */
 265         poll_in_progress = 1;
 266         while (!list_empty(&actor_list)) {
 267                 thread = list_entry(actor_list.next, struct actor, list);
 268                 list_del_init(&thread->list);
 269
 270                 if (thread->state != ACTOR_SCHEDULED)
 271                         log_error("actor_list: thread state corrupted! "
 272                                   "Thread with state %d in actor list.",
 273                                   thread->state);
 274                 thread->state = ACTOR_NOTSCHEDULED;
 275                 log_debug(7, "exec thread %08lx callback", (long)thread);
 276                 thread->callback(thread->data);
 277                 log_debug(7, "thread removed\n");
 278         }
 279         poll_in_progress = 0;
 280
 281         while (!list_empty(&poll_list)) {
 282                 thread = list_entry(poll_list.next, struct actor, list);
 283                 list_del_init(&thread->list);
 284
 285                 if (thread->state != ACTOR_POLL_WAITING)
 286                         log_error("poll_list: thread state corrupted!"
 287                                   "Thread with state %d in poll list.",
 288                                   thread->state);
 289                 thread->state = ACTOR_SCHEDULED;
 290                 list_add_tail(&thread->list, &actor_list);
 291                 log_debug(7, "thread %08lx removed from poll_list",
 292                         (long)thread);
 293         }
 294
 295         ACTOR_TICKS++;
 296 }