rts/Messages.c

   1 /* ---------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 2010
   4  *
   5  * Inter-Capability message passing
   6  *
   7  * --------------------------------------------------------------------------*/
   8
   9 #include "Rts.h"
  10 #include "Messages.h"
  11 #include "Trace.h"
  12 #include "Capability.h"
  13 #include "Schedule.h"
  14 #include "Threads.h"
  15 #include "RaiseAsync.h"
  16 #include "sm/Storage.h"
  17
  18 /* ----------------------------------------------------------------------------
  19    Send a message to another Capability
  20    ------------------------------------------------------------------------- */
  21
  22 #ifdef THREADED_RTS
  23
  24 void sendMessage(Capability *from_cap, Capability *to_cap, Message *msg)
  25 {
  26     ACQUIRE_LOCK(&to_cap->lock);
  27
  28 #ifdef DEBUG
  29     {
  30         const StgInfoTable *i = msg->header.info;
  31         if (i != &stg_MSG_THROWTO_info &&
  32             i != &stg_MSG_BLACKHOLE_info &&
  33             i != &stg_MSG_TRY_WAKEUP_info &&
  34             i != &stg_IND_info && // can happen if a MSG_BLACKHOLE is revoked
  35             i != &stg_WHITEHOLE_info) {
  36             barf("sendMessage: %p", i);
  37         }
  38     }
  39 #endif
  40
  41     msg->link = to_cap->inbox;
  42     to_cap->inbox = msg;
  43
  44     recordClosureMutated(from_cap,(StgClosure*)msg);
  45
  46     if (to_cap->running_task == NULL) {
  47         to_cap->running_task = myTask();
  48             // precond for releaseCapability_()
  49         releaseCapability_(to_cap,rtsFalse);
  50     } else {
  51         contextSwitchCapability(to_cap);
  52     }
  53
  54     RELEASE_LOCK(&to_cap->lock);
  55 }
  56
  57 #endif /* THREADED_RTS */
  58
  59 /* ----------------------------------------------------------------------------
  60    Handle a message
  61    ------------------------------------------------------------------------- */
  62
  63 #ifdef THREADED_RTS
  64
  65 void
  66 executeMessage (Capability *cap, Message *m)
  67 {
  68     const StgInfoTable *i;
  69
  70 loop:
  71     write_barrier(); // allow m->header to be modified by another thread
  72     i = m->header.info;
  73     if (i == &stg_MSG_TRY_WAKEUP_info)
  74     {
  75         StgTSO *tso = ((MessageWakeup *)m)->tso;
  76         debugTraceCap(DEBUG_sched, cap, "message: try wakeup thread %ld",
  77                       (lnat)tso->id);
  78         tryWakeupThread(cap, tso);
  79     }
  80     else if (i == &stg_MSG_THROWTO_info)
  81     {
  82         MessageThrowTo *t = (MessageThrowTo *)m;
  83         nat r;
  84         const StgInfoTable *i;
  85
  86         i = lockClosure((StgClosure*)m);
  87         if (i != &stg_MSG_THROWTO_info) {
  88             unlockClosure((StgClosure*)m, i);
  89             goto loop;
  90         }
  91
  92         debugTraceCap(DEBUG_sched, cap, "message: throwTo %ld -> %ld",
  93                       (lnat)t->source->id, (lnat)t->target->id);
  94
  95         ASSERT(t->source->why_blocked == BlockedOnMsgThrowTo);
  96         ASSERT(t->source->block_info.closure == (StgClosure *)m);
  97
  98         r = throwToMsg(cap, t);
  99
 100         switch (r) {
 101         case THROWTO_SUCCESS:
 102             // this message is done
 103             unlockClosure((StgClosure*)m, &stg_MSG_NULL_info);
 104             tryWakeupThread(cap, t->source);
 105             break;
 106         case THROWTO_BLOCKED:
 107             // unlock the message
 108             unlockClosure((StgClosure*)m, &stg_MSG_THROWTO_info);
 109             break;
 110         }
 111     }
 112     else if (i == &stg_MSG_BLACKHOLE_info)
 113     {
 114         nat r;
 115         MessageBlackHole *b = (MessageBlackHole*)m;
 116
 117         r = messageBlackHole(cap, b);
 118         if (r == 0) {
 119             tryWakeupThread(cap, b->tso);
 120         }
 121         return;
 122     }
 123     else if (i == &stg_IND_info || i == &stg_MSG_NULL_info)
 124     {
 125         // message was revoked
 126         return;
 127     }
 128     else if (i == &stg_WHITEHOLE_info)
 129     {
 130         goto loop;
 131     }
 132     else
 133     {
 134         barf("executeMessage: %p", i);
 135     }
 136 }
 137
 138 #endif
 139
 140 /* ----------------------------------------------------------------------------
 141    Handle a MSG_BLACKHOLE message
 142
 143    This is called from two places: either we just entered a BLACKHOLE
 144    (stg_BLACKHOLE_info), or we received a MSG_BLACKHOLE in our
 145    cap->inbox.
 146
 147    We need to establish whether the BLACKHOLE belongs to
 148    this Capability, and
 149      - if so, arrange to block the current thread on it
 150      - otherwise, forward the message to the right place
 151
 152    Returns:
 153      - 0 if the blocked thread can be woken up by the caller
 154      - 1 if the thread is still blocked, and we promise to send a MSG_TRY_WAKEUP
 155        at some point in the future.
 156
 157    ------------------------------------------------------------------------- */
 158
 159 nat messageBlackHole(Capability *cap, MessageBlackHole *msg)
 160 {
 161     const StgInfoTable *info;
 162     StgClosure *p;
 163     StgBlockingQueue *bq;
 164     StgClosure *bh = msg->bh;
 165     StgTSO *owner;
 166
 167     debugTraceCap(DEBUG_sched, cap, "message: thread %d blocking on blackhole %p",
 168                   (lnat)msg->tso->id, msg->bh);
 169
 170     info = bh->header.info;
 171
 172     // If we got this message in our inbox, it might be that the
 173     // BLACKHOLE has already been updated, and GC has shorted out the
 174     // indirection, so the pointer no longer points to a BLACKHOLE at
 175     // all.
 176     if (info != &stg_BLACKHOLE_info &&
 177         info != &stg_CAF_BLACKHOLE_info &&
 178         info != &stg_WHITEHOLE_info) {
 179         // if it is a WHITEHOLE, then a thread is in the process of
 180         // trying to BLACKHOLE it.  But we know that it was once a
 181         // BLACKHOLE, so there is at least a valid pointer in the
 182         // payload, so we can carry on.
 183         return 0;
 184     }
 185
 186     // The blackhole must indirect to a TSO, a BLOCKING_QUEUE, an IND,
 187     // or a value.
 188 loop:
 189     // NB. VOLATILE_LOAD(), because otherwise gcc hoists the load
 190     // and turns this into an infinite loop.
 191     p = UNTAG_CLOSURE((StgClosure*)VOLATILE_LOAD(&((StgInd*)bh)->indirectee));
 192     info = p->header.info;
 193
 194     if (info == &stg_IND_info)
 195     {
 196         // This could happen, if e.g. we got a BLOCKING_QUEUE that has
 197         // just been replaced with an IND by another thread in
 198         // updateThunk().  In which case, if we read the indirectee
 199         // again we should get the value.
 200         goto loop;
 201     }
 202
 203     else if (info == &stg_TSO_info)
 204     {
 205         owner = deRefTSO((StgTSO *)p);
 206
 207 #ifdef THREADED_RTS
 208         if (owner->cap != cap) {
 209             sendMessage(cap, owner->cap, (Message*)msg);
 210             debugTraceCap(DEBUG_sched, cap, "forwarding message to cap %d", owner->cap->no);
 211             return 1;
 212         }
 213 #endif
 214         // owner is the owner of the BLACKHOLE, and resides on this
 215         // Capability.  msg->tso is the first thread to block on this
 216         // BLACKHOLE, so we first create a BLOCKING_QUEUE object.
 217
 218         bq = (StgBlockingQueue*)allocate(cap, sizeofW(StgBlockingQueue));
 219
 220         // initialise the BLOCKING_QUEUE object
 221         SET_HDR(bq, &stg_BLOCKING_QUEUE_DIRTY_info, CCS_SYSTEM);
 222         bq->bh = bh;
 223         bq->queue = msg;
 224         bq->owner = owner;
 225
 226         msg->link = (MessageBlackHole*)END_TSO_QUEUE;
 227
 228         // All BLOCKING_QUEUES are linked in a list on owner->bq, so
 229         // that we can search through them in the event that there is
 230         // a collision to update a BLACKHOLE and a BLOCKING_QUEUE
 231         // becomes orphaned (see updateThunk()).
 232         bq->link = owner->bq;
 233         owner->bq = bq;
 234         dirty_TSO(cap, owner); // we modified owner->bq
 235
 236         // If the owner of the blackhole is currently runnable, then
 237         // bump it to the front of the run queue.  This gives the
 238         // blocked-on thread a little boost which should help unblock
 239         // this thread, and may avoid a pile-up of other threads
 240         // becoming blocked on the same BLACKHOLE (#3838).
 241         //
 242         // NB. we check to make sure that the owner is not the same as
 243         // the current thread, since in that case it will not be on
 244         // the run queue.
 245         if (owner->why_blocked == NotBlocked && owner->id != msg->tso->id) {
 246             removeFromRunQueue(cap, owner);
 247             pushOnRunQueue(cap,owner);
 248         }
 249
 250         // point to the BLOCKING_QUEUE from the BLACKHOLE
 251         write_barrier(); // make the BQ visible
 252         ((StgInd*)bh)->indirectee = (StgClosure *)bq;
 253         recordClosureMutated(cap,bh); // bh was mutated
 254
 255         debugTraceCap(DEBUG_sched, cap, "thread %d blocked on thread %d",
 256                       (lnat)msg->tso->id, (lnat)owner->id);
 257
 258         return 1; // blocked
 259     }
 260     else if (info == &stg_BLOCKING_QUEUE_CLEAN_info ||
 261              info == &stg_BLOCKING_QUEUE_DIRTY_info)
 262     {
 263         StgBlockingQueue *bq = (StgBlockingQueue *)p;
 264
 265         ASSERT(bq->bh == bh);
 266
 267         owner = deRefTSO(bq->owner);
 268
 269         ASSERT(owner != END_TSO_QUEUE);
 270
 271 #ifdef THREADED_RTS
 272         if (owner->cap != cap) {
 273             sendMessage(cap, owner->cap, (Message*)msg);
 274             debugTraceCap(DEBUG_sched, cap, "forwarding message to cap %d", owner->cap->no);
 275             return 1;
 276         }
 277 #endif
 278
 279         msg->link = bq->queue;
 280         bq->queue = msg;
 281         recordClosureMutated(cap,(StgClosure*)msg);
 282
 283         if (info == &stg_BLOCKING_QUEUE_CLEAN_info) {
 284             bq->header.info = &stg_BLOCKING_QUEUE_DIRTY_info;
 285             recordClosureMutated(cap,(StgClosure*)bq);
 286         }
 287
 288         debugTraceCap(DEBUG_sched, cap, "thread %d blocked on thread %d",
 289                       (lnat)msg->tso->id, (lnat)owner->id);
 290
 291         // See above, #3838
 292         if (owner->why_blocked == NotBlocked && owner->id != msg->tso->id) {
 293             removeFromRunQueue(cap, owner);
 294             pushOnRunQueue(cap,owner);
 295         }
 296
 297         return 1; // blocked
 298     }
 299
 300     return 0; // not blocked
 301 }
 302