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 = UNTAG_CLOSURE(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_EAGER_BLACKHOLE_info &&
 179         info != &stg_WHITEHOLE_info) {
 180         // if it is a WHITEHOLE, then a thread is in the process of
 181         // trying to BLACKHOLE it.  But we know that it was once a
 182         // BLACKHOLE, so there is at least a valid pointer in the
 183         // payload, so we can carry on.
 184         return 0;
 185     }
 186
 187     // The blackhole must indirect to a TSO, a BLOCKING_QUEUE, an IND,
 188     // or a value.
 189 loop:
 190     // NB. VOLATILE_LOAD(), because otherwise gcc hoists the load
 191     // and turns this into an infinite loop.
 192     p = UNTAG_CLOSURE((StgClosure*)VOLATILE_LOAD(&((StgInd*)bh)->indirectee));
 193     info = p->header.info;
 194
 195     if (info == &stg_IND_info)
 196     {
 197         // This could happen, if e.g. we got a BLOCKING_QUEUE that has
 198         // just been replaced with an IND by another thread in
 199         // updateThunk().  In which case, if we read the indirectee
 200         // again we should get the value.
 201         goto loop;
 202     }
 203
 204     else if (info == &stg_TSO_info)
 205     {
 206         owner = deRefTSO((StgTSO *)p);
 207
 208 #ifdef THREADED_RTS
 209         if (owner->cap != cap) {
 210             sendMessage(cap, owner->cap, (Message*)msg);
 211             debugTraceCap(DEBUG_sched, cap, "forwarding message to cap %d", owner->cap->no);
 212             return 1;
 213         }
 214 #endif
 215         // owner is the owner of the BLACKHOLE, and resides on this
 216         // Capability.  msg->tso is the first thread to block on this
 217         // BLACKHOLE, so we first create a BLOCKING_QUEUE object.
 218
 219         bq = (StgBlockingQueue*)allocate(cap, sizeofW(StgBlockingQueue));
 220
 221         // initialise the BLOCKING_QUEUE object
 222         SET_HDR(bq, &stg_BLOCKING_QUEUE_DIRTY_info, CCS_SYSTEM);
 223         bq->bh = bh;
 224         bq->queue = msg;
 225         bq->owner = owner;
 226
 227         msg->link = (MessageBlackHole*)END_TSO_QUEUE;
 228
 229         // All BLOCKING_QUEUES are linked in a list on owner->bq, so
 230         // that we can search through them in the event that there is
 231         // a collision to update a BLACKHOLE and a BLOCKING_QUEUE
 232         // becomes orphaned (see updateThunk()).
 233         bq->link = owner->bq;
 234         owner->bq = bq;
 235         dirty_TSO(cap, owner); // we modified owner->bq
 236
 237         // If the owner of the blackhole is currently runnable, then
 238         // bump it to the front of the run queue.  This gives the
 239         // blocked-on thread a little boost which should help unblock
 240         // this thread, and may avoid a pile-up of other threads
 241         // becoming blocked on the same BLACKHOLE (#3838).
 242         //
 243         // NB. we check to make sure that the owner is not the same as
 244         // the current thread, since in that case it will not be on
 245         // the run queue.
 246         if (owner->why_blocked == NotBlocked && owner->id != msg->tso->id) {
 247             removeFromRunQueue(cap, owner);
 248             pushOnRunQueue(cap,owner);
 249         }
 250
 251         // point to the BLOCKING_QUEUE from the BLACKHOLE
 252         write_barrier(); // make the BQ visible
 253         ((StgInd*)bh)->indirectee = (StgClosure *)bq;
 254         recordClosureMutated(cap,bh); // bh was mutated
 255
 256         debugTraceCap(DEBUG_sched, cap, "thread %d blocked on thread %d",
 257                       (lnat)msg->tso->id, (lnat)owner->id);
 258
 259         return 1; // blocked
 260     }
 261     else if (info == &stg_BLOCKING_QUEUE_CLEAN_info ||
 262              info == &stg_BLOCKING_QUEUE_DIRTY_info)
 263     {
 264         StgBlockingQueue *bq = (StgBlockingQueue *)p;
 265
 266         ASSERT(bq->bh == bh);
 267
 268         owner = deRefTSO(bq->owner);
 269
 270         ASSERT(owner != END_TSO_QUEUE);
 271
 272 #ifdef THREADED_RTS
 273         if (owner->cap != cap) {
 274             sendMessage(cap, owner->cap, (Message*)msg);
 275             debugTraceCap(DEBUG_sched, cap, "forwarding message to cap %d", owner->cap->no);
 276             return 1;
 277         }
 278 #endif
 279
 280         msg->link = bq->queue;
 281         bq->queue = msg;
 282         recordClosureMutated(cap,(StgClosure*)msg);
 283
 284         if (info == &stg_BLOCKING_QUEUE_CLEAN_info) {
 285             bq->header.info = &stg_BLOCKING_QUEUE_DIRTY_info;
 286             recordClosureMutated(cap,(StgClosure*)bq);
 287         }
 288
 289         debugTraceCap(DEBUG_sched, cap, "thread %d blocked on thread %d",
 290                       (lnat)msg->tso->id, (lnat)owner->id);
 291
 292         // See above, #3838
 293         if (owner->why_blocked == NotBlocked && owner->id != msg->tso->id) {
 294             removeFromRunQueue(cap, owner);
 295             pushOnRunQueue(cap,owner);
 296         }
 297
 298         return 1; // blocked
 299     }
 300
 301     return 0; // not blocked
 302 }
 303