rx-clock-backward-detection-20080317
[openafs.git] / src / rx / rx_event.c
1 /*
2  * Copyright 2000, International Business Machines Corporation and others.
3  * All Rights Reserved.
4  * 
5  * This software has been released under the terms of the IBM Public
6  * License.  For details, see the LICENSE file in the top-level source
7  * directory or online at http://www.openafs.org/dl/license10.html
8  */
9
10 #include <afsconfig.h>
11 #ifdef  KERNEL
12 #include "afs/param.h"
13 #else
14 #include <afs/param.h>
15 #endif
16
17 #ifdef AFS_SUN59_ENV
18 #include <sys/time_impl.h>
19 #endif
20
21 RCSID
22     ("$Header$");
23
24 #ifdef KERNEL
25 #ifndef UKERNEL
26 #include "afs/afs_osi.h"
27 #else /* !UKERNEL */
28 #include "afs/sysincludes.h"
29 #include "afsincludes.h"
30 #endif /* !UKERNEL */
31 #include "rx/rx_clock.h"
32 #include "rx/rx_queue.h"
33 #include "rx/rx_event.h"
34 #include "rx/rx_kernel.h"
35 #include "rx_kmutex.h"
36 #ifdef RX_ENABLE_LOCKS
37 #include "rx/rx.h"
38 #endif /* RX_ENABLE_LOCKS */
39 #include "rx/rx_globals.h"
40 #if defined(AFS_SGI_ENV)
41 #include "sys/debug.h"
42 /* These are necessary to get curproc (used by GLOCK asserts) to work. */
43 #include "h/proc.h"
44 #if !defined(AFS_SGI64_ENV) && !defined(UKERNEL)
45 #include "h/user.h"
46 #endif
47 extern void *osi_Alloc();
48 #endif
49 #if defined(AFS_OBSD_ENV)
50 #include "h/proc.h"
51 #endif
52 #else /* KERNEL */
53 #include <stdio.h>
54 #include "rx_clock.h"
55 #include "rx_queue.h"
56 #include "rx_event.h"
57 #include "rx_user.h"
58 #ifdef AFS_PTHREAD_ENV
59 #include <rx/rx_pthread.h>
60 #else
61 #include "rx_lwp.h"
62 #endif
63 #ifdef RX_ENABLE_LOCKS
64 #include "rx.h"
65 #endif /* RX_ENABLE_LOCKS */
66 #include "rx_globals.h"
67 #ifdef AFS_NT40_ENV
68 #include <malloc.h>
69 #endif
70 #endif /* KERNEL */
71
72
73 /* All event processing is relative to the apparent current time given by clock_GetTime */
74
75 /* This should be static, but event_test wants to look at the free list... */
76 struct rx_queue rxevent_free;   /* It's somewhat bogus to use a doubly-linked queue for the free list */
77 struct rx_queue rxepoch_free;   /* It's somewhat bogus to use a doubly-linked queue for the free list */
78 static struct rx_queue rxepoch_queue;   /* list of waiting epochs */
79 static int rxevent_allocUnit = 10;      /* Allocation unit (number of event records to allocate at one time) */
80 static int rxepoch_allocUnit = 10;      /* Allocation unit (number of epoch records to allocate at one time) */
81 int rxevent_nFree;              /* Number of free event records */
82 int rxevent_nPosted;            /* Current number of posted events */
83 int rxepoch_nFree;              /* Number of free epoch records */
84 static void (*rxevent_ScheduledEarlierEvent) (void);    /* Proc to call when an event is scheduled that is earlier than all other events */
85 struct xfreelist {
86     void *mem;
87     int size;
88     struct xfreelist *next;
89 };
90 static struct xfreelist *xfreemallocs = 0, *xsp = 0;
91
92 struct clock rxevent_nextRaiseEvents;   /* Time of next call to raise events */
93 struct clock rxevent_lastEvent;        /* backwards time detection */
94 int rxevent_raiseScheduled;     /* true if raise events is scheduled */
95
96 #ifdef RX_ENABLE_LOCKS
97 #ifdef RX_LOCKS_DB
98 /* rxdb_fileID is used to identify the lock location, along with line#. */
99 static int rxdb_fileID = RXDB_FILE_RX_EVENT;
100 #endif /* RX_LOCKS_DB */
101 #define RX_ENABLE_LOCKS  1
102 afs_kmutex_t rxevent_lock;
103 #endif /* RX_ENABLE_LOCKS */
104
105 #ifdef AFS_PTHREAD_ENV
106 /*
107  * This mutex protects the following global variables:
108  * rxevent_initialized
109  */
110
111 #include <assert.h>
112 pthread_mutex_t rx_event_mutex;
113 #define LOCK_EV_INIT assert(pthread_mutex_lock(&rx_event_mutex)==0)
114 #define UNLOCK_EV_INIT assert(pthread_mutex_unlock(&rx_event_mutex)==0)
115 #else
116 #define LOCK_EV_INIT
117 #define UNLOCK_EV_INIT
118 #endif /* AFS_PTHREAD_ENV */
119
120
121 int
122 rxevent_adjTimes(struct clock *adjTime)
123 {
124     /* backwards clock correction */
125     int nAdjusted = 0;
126     struct rxepoch *qep, *nqep;
127     struct rxevent *qev, *nqev;
128     
129     for (queue_Scan(&rxepoch_queue, qep, nqep, rxepoch)) {
130         for (queue_Scan(&qep->events, qev, nqev, rxevent)) {
131             if (clock_Gt(&qev->eventTime, adjTime)) {
132                 clock_Sub(&qev->eventTime, adjTime); 
133                 nAdjusted++;
134             }
135         }
136         if (qep->epochSec > adjTime->sec) {
137             qep->epochSec -= adjTime->sec;
138         }
139     }
140     return nAdjusted;
141 }
142
143 /* Pass in the number of events to allocate at a time */
144 int rxevent_initialized = 0;
145 void
146 rxevent_Init(int nEvents, void (*scheduler) (void))
147 {
148     LOCK_EV_INIT;
149     if (rxevent_initialized) {
150         UNLOCK_EV_INIT;
151         return;
152     }
153     MUTEX_INIT(&rxevent_lock, "rxevent_lock", MUTEX_DEFAULT, 0);
154     clock_Init();
155     if (nEvents)
156         rxevent_allocUnit = nEvents;
157     queue_Init(&rxevent_free);
158     queue_Init(&rxepoch_free);
159     queue_Init(&rxepoch_queue);
160     rxevent_nFree = rxevent_nPosted = 0;
161     rxepoch_nFree = 0;
162     rxevent_ScheduledEarlierEvent = scheduler;
163     rxevent_initialized = 1;
164     clock_Zero(&rxevent_nextRaiseEvents);
165     clock_Zero(&rxevent_lastEvent);
166     rxevent_raiseScheduled = 0;
167     UNLOCK_EV_INIT;
168 }
169
170 /* Create and initialize new epoch structure */
171 struct rxepoch *
172 rxepoch_Allocate(struct clock *when)
173 {
174     struct rxepoch *ep;
175     int i;
176
177     /* If we are short on free epoch entries, create a block of new oned
178      * and add them to the free queue */
179     if (queue_IsEmpty(&rxepoch_free)) {
180 #if    defined(AFS_AIX32_ENV) && defined(KERNEL)
181         ep = (struct rxepoch *)rxi_Alloc(sizeof(struct rxepoch));
182         queue_Append(&rxepoch_free, &ep[0]), rxepoch_nFree++;
183 #else
184         ep = (struct rxepoch *)
185             osi_Alloc(sizeof(struct rxepoch) * rxepoch_allocUnit);
186         xsp = xfreemallocs;
187         xfreemallocs =
188             (struct xfreelist *)osi_Alloc(sizeof(struct xfreelist));
189         xfreemallocs->mem = (void *)ep;
190         xfreemallocs->size = sizeof(struct rxepoch) * rxepoch_allocUnit;
191         xfreemallocs->next = xsp;
192         for (i = 0; i < rxepoch_allocUnit; i++)
193             queue_Append(&rxepoch_free, &ep[i]), rxepoch_nFree++;
194 #endif
195     }
196     ep = queue_First(&rxepoch_free, rxepoch);
197     queue_Remove(ep);
198     rxepoch_nFree--;
199     ep->epochSec = when->sec;
200     queue_Init(&ep->events);
201     return ep;
202 }
203
204 /* Add the indicated event (function, arg) at the specified clock time.  The
205  * "when" argument specifies when "func" should be called, in clock (clock.h)
206  * units. */
207
208 static struct rxevent *
209 _rxevent_Post(struct clock *when, struct clock *now, void (*func) (), 
210               void *arg, void *arg1, int arg2, int newargs)
211 {
212     register struct rxevent *ev, *evqe, *evqpr;
213     register struct rxepoch *ep, *epqe, *epqpr;
214     int isEarliest = 0;
215
216     MUTEX_ENTER(&rxevent_lock);
217 #ifdef RXDEBUG
218     if (rx_Log_event) {
219         struct clock now1;
220         clock_GetTime(&now1);
221         fprintf(rx_Log_event, "%d.%d: rxevent_Post(%d.%d, %lp, %lp, %lp, %d)\n",
222                 (int)now1.sec, (int)now1.usec, (int)when->sec, (int)when->usec,
223                 func, arg,
224                 arg1, arg2);
225     }
226 #endif
227     /* If a time was provided, check for consistency */
228     if (now->sec) {
229         if (clock_Gt(&rxevent_lastEvent, now)) {
230             struct clock adjTime = rxevent_lastEvent;
231             clock_Sub(&adjTime, now);
232             rxevent_adjTimes(&adjTime);
233         }
234         rxevent_lastEvent = *now;
235     }
236     /* Get a pointer to the epoch for this event, if none is found then
237      * create a new epoch and insert it into the sorted list */
238     for (ep = NULL, queue_ScanBackwards(&rxepoch_queue, epqe, epqpr, rxepoch)) {
239         if (when->sec == epqe->epochSec) {
240             /* already have an structure for this epoch */
241             ep = epqe;
242             if (ep == queue_First(&rxepoch_queue, rxepoch))
243                 isEarliest = 1;
244             break;
245         } else if (when->sec > epqe->epochSec) {
246             /* Create a new epoch and insert after qe */
247             ep = rxepoch_Allocate(when);
248             queue_InsertAfter(epqe, ep);
249             break;
250         }
251     }
252     if (ep == NULL) {
253         /* Create a new epoch and place it at the head of the list */
254         ep = rxepoch_Allocate(when);
255         queue_Prepend(&rxepoch_queue, ep);
256         isEarliest = 1;
257     }
258
259     /* If we're short on free event entries, create a block of new ones and add
260      * them to the free queue */
261     if (queue_IsEmpty(&rxevent_free)) {
262         register int i;
263 #if     defined(AFS_AIX32_ENV) && defined(KERNEL)
264         ev = (struct rxevent *)rxi_Alloc(sizeof(struct rxevent));
265         queue_Append(&rxevent_free, &ev[0]), rxevent_nFree++;
266 #else
267         ev = (struct rxevent *)osi_Alloc(sizeof(struct rxevent) *
268                                          rxevent_allocUnit);
269         xsp = xfreemallocs;
270         xfreemallocs =
271             (struct xfreelist *)osi_Alloc(sizeof(struct xfreelist));
272         xfreemallocs->mem = (void *)ev;
273         xfreemallocs->size = sizeof(struct rxevent) * rxevent_allocUnit;
274         xfreemallocs->next = xsp;
275         for (i = 0; i < rxevent_allocUnit; i++)
276             queue_Append(&rxevent_free, &ev[i]), rxevent_nFree++;
277 #endif
278     }
279
280     /* Grab and initialize a new rxevent structure */
281     ev = queue_First(&rxevent_free, rxevent);
282     queue_Remove(ev);
283     rxevent_nFree--;
284
285     /* Record user defined event state */
286     ev->eventTime = *when;
287     ev->func = func;
288     ev->arg = arg;
289     ev->arg1 = arg1;
290     ev->arg2 = arg2;
291     ev->newargs = newargs;
292     rxevent_nPosted += 1;       /* Rather than ++, to shut high-C up
293                                  *  regarding never-set variables
294                                  */
295
296     /* Insert the event into the sorted list of events for this epoch */
297     for (queue_ScanBackwards(&ep->events, evqe, evqpr, rxevent)) {
298         if (when->usec >= evqe->eventTime.usec) {
299             /* Insert event after evqe */
300             queue_InsertAfter(evqe, ev);
301             MUTEX_EXIT(&rxevent_lock);
302             return ev;
303         }
304     }
305     /* Insert event at head of current epoch */
306     queue_Prepend(&ep->events, ev);
307     if (isEarliest && rxevent_ScheduledEarlierEvent
308         && (!rxevent_raiseScheduled
309             || clock_Lt(&ev->eventTime, &rxevent_nextRaiseEvents))) {
310         rxevent_raiseScheduled = 1;
311         clock_Zero(&rxevent_nextRaiseEvents);
312         MUTEX_EXIT(&rxevent_lock);
313         /* Notify our external scheduler */
314         (*rxevent_ScheduledEarlierEvent) ();
315         MUTEX_ENTER(&rxevent_lock);
316     }
317     MUTEX_EXIT(&rxevent_lock);
318     return ev;
319 }
320
321 struct rxevent *
322 rxevent_Post(struct clock *when, void (*func) (), void *arg, void *arg1)
323 {
324     struct clock now;
325     clock_Zero(&now);
326     return _rxevent_Post(when, &now, func, arg, arg1, 0, 0);
327 }
328
329 struct rxevent *
330 rxevent_Post2(struct clock *when, void (*func) (), void *arg, void *arg1,
331               int arg2)
332 {
333     struct clock now;
334     clock_Zero(&now);
335     return _rxevent_Post(when, &now, func, arg, arg1, arg2, 1);
336 }
337
338 struct rxevent *
339 rxevent_PostNow(struct clock *when, struct clock *now, void (*func) (), 
340                 void *arg, void *arg1)
341 {
342     return _rxevent_Post(when, now, func, arg, arg1, 0, 0);
343 }
344
345 struct rxevent *
346 rxevent_PostNow2(struct clock *when, struct clock *now, void (*func) (), 
347                  void *arg, void *arg1, int arg2)
348 {
349     return _rxevent_Post(when, now, func, arg, arg1, arg2, 1);
350 }
351
352 /* Cancel an event by moving it from the event queue to the free list.
353  * Warning, the event must be on the event queue!  If not, this should core
354  * dump (reference through 0).  This routine should be called using the macro
355  * event_Cancel, which checks for a null event and also nulls the caller's
356  * event pointer after cancelling the event.
357  */
358 #ifdef RX_ENABLE_LOCKS
359 #ifdef RX_REFCOUNT_CHECK
360 int rxevent_Cancel_type = 0;
361 #endif
362 #endif
363
364 void
365 rxevent_Cancel_1(register struct rxevent *ev, register struct rx_call *call,
366                  register int type)
367 {
368 #ifdef RXDEBUG
369     if (rx_Log_event) {
370         struct clock now;
371         clock_GetTime(&now);
372         fprintf(rx_Log_event, "%d.%d: rxevent_Cancel_1(%d.%d, %lp, %lp)\n",
373                 (int)now.sec, (int)now.usec, (int)ev->eventTime.sec,
374                 (int)ev->eventTime.usec, ev->func,
375                 ev->arg);
376     }
377 #endif
378     /* Append it to the free list (rather than prepending) to keep the free
379      * list hot so nothing pages out
380      */
381     MUTEX_ENTER(&rxevent_lock);
382     if (!ev) {
383         MUTEX_EXIT(&rxevent_lock);
384         return;
385     }
386 #ifdef RX_ENABLE_LOCKS
387     /* It's possible we're currently processing this event. */
388     if (queue_IsOnQueue(ev)) {
389         queue_MoveAppend(&rxevent_free, ev);
390         rxevent_nPosted--;
391         rxevent_nFree++;
392         if (call) {
393             call->refCount--;
394 #ifdef RX_REFCOUNT_CHECK
395             call->refCDebug[type]--;
396             if (call->refCDebug[type] < 0) {
397                 rxevent_Cancel_type = type;
398                 osi_Panic("rxevent_Cancel: call refCount < 0");
399             }
400 #endif /* RX_REFCOUNT_CHECK */
401         }
402     }
403 #else /* RX_ENABLE_LOCKS */
404     queue_MoveAppend(&rxevent_free, ev);
405     rxevent_nPosted--;
406     rxevent_nFree++;
407 #endif /* RX_ENABLE_LOCKS */
408     MUTEX_EXIT(&rxevent_lock);
409 }
410
411 /* Process all epochs that have expired relative to the current clock time
412  * (which is not re-evaluated unless clock_NewTime has been called).  The
413  * relative time to the next epoch is returned in the output parameter next
414  * and the function returns 1.  If there are is no next epoch, the function
415  * returns 0.
416  */
417 int
418 rxevent_RaiseEvents(struct clock *next)
419 {
420     register struct rxepoch *ep;
421     register struct rxevent *ev;
422     volatile struct clock now;
423     MUTEX_ENTER(&rxevent_lock);
424
425     /* Events are sorted by time, so only scan until an event is found that has
426      * not yet timed out */
427
428     clock_Zero(&now);
429     while (queue_IsNotEmpty(&rxepoch_queue)) {
430         ep = queue_First(&rxepoch_queue, rxepoch);
431         if (queue_IsEmpty(&ep->events)) {
432             queue_Remove(ep);
433             queue_Append(&rxepoch_free, ep);
434             rxepoch_nFree++;
435             continue;
436         }
437         do {
438         reraise:
439             ev = queue_First(&ep->events, rxevent);
440             if (clock_Lt(&now, &ev->eventTime)) {
441                 clock_GetTime(&now);
442                 if (clock_Gt(&rxevent_lastEvent, &now)) {
443                     struct clock adjTime = rxevent_lastEvent;
444                     int adjusted;
445                     clock_Sub(&adjTime, &now);
446                     adjusted = rxevent_adjTimes(&adjTime);
447                     rxevent_lastEvent = now;
448                     if (adjusted > 0)
449                         goto reraise;
450                 }
451                 if (clock_Lt(&now, &ev->eventTime)) {
452                     *next = rxevent_nextRaiseEvents = ev->eventTime;
453                     rxevent_raiseScheduled = 1;
454                     clock_Sub(next, &now);
455                     MUTEX_EXIT(&rxevent_lock);
456                     return 1;
457                 }
458             }
459             queue_Remove(ev);
460             rxevent_nPosted--;
461             MUTEX_EXIT(&rxevent_lock);
462             if (ev->newargs) {
463                 ev->func(ev, ev->arg, ev->arg1, ev->arg2);
464             } else {
465                 ev->func(ev, ev->arg, ev->arg1);
466             }
467             MUTEX_ENTER(&rxevent_lock);
468             queue_Append(&rxevent_free, ev);
469             rxevent_nFree++;
470         } while (queue_IsNotEmpty(&ep->events));
471     }
472 #ifdef RXDEBUG
473     if (rx_Log_event)
474         fprintf(rx_Log_event, "rxevent_RaiseEvents(%d.%d)\n", (int)now.sec,
475                 (int)now.usec);
476 #endif
477     rxevent_raiseScheduled = 0;
478     MUTEX_EXIT(&rxevent_lock);
479     return 0;
480 }
481
482 void
483 shutdown_rxevent(void)
484 {
485     struct xfreelist *xp, *nxp;
486
487     LOCK_EV_INIT;
488     if (!rxevent_initialized) {
489         UNLOCK_EV_INIT;
490         return;
491     }
492     rxevent_initialized = 0;
493     UNLOCK_EV_INIT;
494     MUTEX_DESTROY(&rxevent_lock);
495 #if     defined(AFS_AIX32_ENV) && defined(KERNEL)
496     /* Everything is freed in afs_osinet.c */
497 #else
498     xp = xfreemallocs;
499     while (xp) {
500         nxp = xp->next;
501         osi_Free((char *)xp->mem, xp->size);
502         osi_Free((char *)xp, sizeof(struct xfreelist));
503         xp = nxp;
504     }
505     xfreemallocs = NULL;
506 #endif
507
508 }