diff options
Diffstat (limited to 'generic/tclIORChan.c')
| -rw-r--r-- | generic/tclIORChan.c | 174 |
1 files changed, 159 insertions, 15 deletions
diff --git a/generic/tclIORChan.c b/generic/tclIORChan.c index 49e2930..76002f6 100644 --- a/generic/tclIORChan.c +++ b/generic/tclIORChan.c @@ -39,6 +39,9 @@ static int ReflectOutput(ClientData clientData, const char *buf, int toWrite, int *errorCodePtr); static void ReflectWatch(ClientData clientData, int mask); static int ReflectBlock(ClientData clientData, int mode); +#ifdef TCL_THREADS +static void ReflectThread(ClientData clientData, int action); +#endif static Tcl_WideInt ReflectSeekWide(ClientData clientData, Tcl_WideInt offset, int mode, int *errorCodePtr); static int ReflectSeek(ClientData clientData, long offset, @@ -71,7 +74,11 @@ static const Tcl_ChannelType tclRChannelType = { NULL, /* Flush channel. Not used by core. NULL'able */ NULL, /* Handle events. NULL'able */ ReflectSeekWide, /* Move access point (64 bit). NULL'able */ +#ifdef TCL_THREADS + ReflectThread, /* thread action, tracking owner */ +#else NULL, /* thread action */ +#endif NULL /* truncate */ }; @@ -89,7 +96,8 @@ typedef struct { * command is gone. */ #ifdef TCL_THREADS - Tcl_ThreadId thread; /* Thread the 'interp' belongs to. */ + Tcl_ThreadId thread; /* Thread the 'interp' belongs to. == Handler thread */ + Tcl_ThreadId owner; /* Thread owning the structure. == Channel thread */ #endif /* See [==] as well. @@ -390,7 +398,7 @@ TCL_DECLARE_MUTEX(rcForwardMutex) * leak resources when threads go away. */ -static void ForwardOpToOwnerThread(ReflectedChannel *rcPtr, +static void ForwardOpToHandlerThread(ReflectedChannel *rcPtr, ForwardedOperation op, const void *param); static int ForwardProc(Tcl_Event *evPtr, int mask); static void SrcExitProc(ClientData clientData); @@ -765,6 +773,48 @@ TclChanCreateObjCmd( *---------------------------------------------------------------------- */ +typedef struct ReflectEvent { + Tcl_Event header; + ReflectedChannel* rcPtr; + int events; +} ReflectEvent; + +static int +ReflectEventRun (Tcl_Event* ev, int flags) +{ + /* OWNER thread + * + * Note: When the channel is closed any pending events of this type are + * deleted. See ReflectClose() for the Tcl_DeleteEvents() calls + * accomplishing that. + */ + + ReflectEvent* e = (ReflectEvent*) ev; + + Tcl_NotifyChannel (e->rcPtr->chan, e->events); + return 1; +} + +static int +ReflectEventDelete (Tcl_Event* ev, ClientData cd) +{ + /* OWNER thread + * + * Invoked by DeleteThreadReflectedChannelMap() and ReflectClose(). The + * latter ensures that no pending events of this type are run on an + * invalid channel. + */ + + ReflectEvent* e = (ReflectEvent*) ev; + + if ((ev->proc != ReflectEventRun) || + ((cd != NULL) && + (cd != e->rcPtr))) { + return 0; + } + return 1; +} + int TclChanPostEventObjCmd( ClientData clientData, @@ -773,6 +823,8 @@ TclChanPostEventObjCmd( Tcl_Obj *const *objv) { /* + * Ensure -> HANDLER thread + * * Syntax: chan postevent CHANNEL EVENTSPEC * [0] [1] [2] [3] * @@ -882,7 +934,38 @@ TclChanPostEventObjCmd( * We have the channel and the events to post. */ - Tcl_NotifyChannel(chan, events); + { + ReflectEvent* ev = ckalloc (sizeof (ReflectEvent)); + ev->header.proc = ReflectEventRun; + ev->events = events; + ev->rcPtr = rcPtr; + + /* + * We are not preserving the structure here. When the channel is + * closed any pending events are deleted, see ReflectClose(), and + * ReflectEventDelete(). Trying to preserve and later release when the + * event is run may generate a situation where the channel structure + * is deleted but not our structure, crashing in + * FreeReflectedChannel(). + */ + + /* Force creation of the RCM, for proper cleanup on thread teardown */ + /* The teardown of unprocessed events is currently coupled to the thread reflected channel map */ + (void) GetThreadReflectedChannelMap (); + + if (rcPtr->owner == rcPtr->thread) { + Tcl_QueueEvent ((Tcl_Event*) ev, TCL_QUEUE_TAIL); + } else { + /* XXX Race condition !! + * XXX The destination thread may not exist anymore already. + * XXX (Delayed postevent executed after channel got removed). + * XXX Can we detect this ? (check the validity of the owner threadid ?) + * XXX Actually, in that case the channel should be dead also ! + */ + Tcl_ThreadQueueEvent (rcPtr->owner, (Tcl_Event*) ev, TCL_QUEUE_TAIL); + Tcl_ThreadAlert (rcPtr->owner); + } + } /* * Squash interp results left by the event script. @@ -1067,9 +1150,12 @@ ReflectClose( if (rcPtr->thread != Tcl_GetCurrentThread()) { ForwardParam p; - ForwardOpToOwnerThread(rcPtr, ForwardedClose, &p); + ForwardOpToHandlerThread(rcPtr, ForwardedClose, &p); result = p.base.code; + /* Now squash the pending reflection events for this channel. */ + Tcl_DeleteEvents (ReflectEventDelete, rcPtr); + if (result != TCL_OK) { FreeReceivedError(&p); } @@ -1100,9 +1186,12 @@ ReflectClose( if (rcPtr->thread != Tcl_GetCurrentThread()) { ForwardParam p; - ForwardOpToOwnerThread(rcPtr, ForwardedClose, &p); + ForwardOpToHandlerThread(rcPtr, ForwardedClose, &p); result = p.base.code; + /* Now squash the pending reflection events for this channel. */ + Tcl_DeleteEvents (ReflectEventDelete, rcPtr); + Tcl_EventuallyFree (rcPtr, (Tcl_FreeProc *) FreeReflectedChannel); if (result != TCL_OK) { @@ -1207,7 +1296,7 @@ ReflectInput( p.input.buf = buf; p.input.toRead = toRead; - ForwardOpToOwnerThread(rcPtr, ForwardedInput, &p); + ForwardOpToHandlerThread(rcPtr, ForwardedInput, &p); if (p.base.code != TCL_OK) { if (p.base.code < 0) { @@ -1322,7 +1411,7 @@ ReflectOutput( p.output.buf = buf; p.output.toWrite = toWrite; - ForwardOpToOwnerThread(rcPtr, ForwardedOutput, &p); + ForwardOpToHandlerThread(rcPtr, ForwardedOutput, &p); if (p.base.code != TCL_OK) { if (p.base.code < 0) { @@ -1438,7 +1527,7 @@ ReflectSeekWide( p.seek.seekMode = seekMode; p.seek.offset = offset; - ForwardOpToOwnerThread(rcPtr, ForwardedSeek, &p); + ForwardOpToHandlerThread(rcPtr, ForwardedSeek, &p); if (p.base.code != TCL_OK) { PassReceivedError(rcPtr->chan, &p); @@ -1562,7 +1651,7 @@ ReflectWatch( ForwardParam p; p.watch.mask = mask; - ForwardOpToOwnerThread(rcPtr, ForwardedWatch, &p); + ForwardOpToHandlerThread(rcPtr, ForwardedWatch, &p); /* * Any failure from the forward is ignored. We have no place to put @@ -1620,7 +1709,7 @@ ReflectBlock( p.block.nonblocking = nonblocking; - ForwardOpToOwnerThread(rcPtr, ForwardedBlock, &p); + ForwardOpToHandlerThread(rcPtr, ForwardedBlock, &p); if (p.base.code != TCL_OK) { PassReceivedError(rcPtr->chan, &p); @@ -1650,6 +1739,42 @@ ReflectBlock( return errorNum; } +#ifdef TCL_THREADS +/* + *---------------------------------------------------------------------- + * + * ReflectThread -- + * + * This function is invoked to tell the channel about thread movements. + * + * Results: + * None. + * + * Side effects: + * Allocates memory. Arbitrary, as it calls upon a script. + * + *---------------------------------------------------------------------- + */ + +static void +ReflectThread(ClientData clientData, int action) +{ + ReflectedChannel *rcPtr = clientData; + + switch (action) { + case TCL_CHANNEL_THREAD_INSERT: + rcPtr->owner = Tcl_GetCurrentThread(); + break; + case TCL_CHANNEL_THREAD_REMOVE: + rcPtr->owner = NULL; + break; + default: + Tcl_Panic ("Unknown thread action code."); + break; + } +} + +#endif /* *---------------------------------------------------------------------- * @@ -1689,7 +1814,7 @@ ReflectSetOption( p.setOpt.name = optionName; p.setOpt.value = newValue; - ForwardOpToOwnerThread(rcPtr, ForwardedSetOpt, &p); + ForwardOpToHandlerThread(rcPtr, ForwardedSetOpt, &p); if (p.base.code != TCL_OK) { Tcl_Obj *err = Tcl_NewStringObj(p.base.msgStr, -1); @@ -1775,7 +1900,7 @@ ReflectGetOption( opcode = ForwardedGetOpt; } - ForwardOpToOwnerThread(rcPtr, opcode, &p); + ForwardOpToHandlerThread(rcPtr, opcode, &p); if (p.base.code != TCL_OK) { Tcl_Obj *err = Tcl_NewStringObj(p.base.msgStr, -1); @@ -2673,6 +2798,15 @@ DeleteThreadReflectedChannelMap( Tcl_MutexUnlock(&rcForwardMutex); /* + * Run over the event queue of this thread and remove all ReflectEvent's + * still pending. These are inbound events for reflected channels this + * thread owns but doesn't handle. The inverse of the channel map + * actually. + */ + + Tcl_DeleteEvents (ReflectEventDelete, NULL); + + /* * Get the map of all channels handled by the current thread. This is a * ReflectedChannelMap, but on a per-thread basis, not per-interp. Go * through the channels, remove all, mark them as dead. @@ -2693,11 +2827,16 @@ DeleteThreadReflectedChannelMap( } static void -ForwardOpToOwnerThread( +ForwardOpToHandlerThread( ReflectedChannel *rcPtr, /* Channel instance */ ForwardedOperation op, /* Forwarded driver operation */ const void *param) /* Arguments */ { + /* + * Core of the communication from OWNER to HANDLER thread. + * The receiver is ForwardProc() below. + */ + Tcl_ThreadId dst = rcPtr->thread; ForwardingEvent *evPtr; ForwardingResult *resultPtr; @@ -2750,7 +2889,7 @@ ForwardOpToOwnerThread( /* * Ensure cleanup of the event if the origin thread exits while this event * is pending or in progress. Exit of the destination thread is handled by - * DeleteThreadReflectionChannelMap(), this is set up by + * DeleteThreadReflectedChannelMap(), this is set up by * GetThreadReflectedChannelMap(). This is what we use the 'forwardList' * (see above) for. */ @@ -2765,7 +2904,7 @@ ForwardOpToOwnerThread( Tcl_ThreadAlert(dst); /* - * (*) Block until the other thread has either processed the transfer or + * (*) Block until the handler thread has either processed the transfer or * rejected it. */ @@ -2813,6 +2952,11 @@ ForwardProc( int mask) { /* + * HANDLER thread. + + * The receiver part for the operations coming from the OWNER thread. + * See ForwardOpToHandlerThread() for the transmitter. + * * Notes regarding access to the referenced data. * * In principle the data belongs to the originating thread (see |