l4/pkg/plr/server/src/thread_group.h

   1 #pragma once
   2
   3 /*
   4  * thread_group.h --
   5  *
   6  *    Thread group -> representation of a single replicated thread
   7  *
   8  * (c) 2012-2013 Björn Döbel <doebel@os.inf.tu-dresden.de>,
   9  *     economic rights: Technische Universität Dresden (Germany)
  10  * This file is part of TUD:OS and distributed under the terms of the
  11  * GNU General Public License 2.
  12  * Please see the COPYING-GPL-2 file for details.
  13  */
  14
  15
  16 #include "app"
  17 #include "app_loading"
  18
  19 #include <l4/sys/ipc.h>
  20
  21 #include <pthread.h>
  22 #include <pthread-l4.h>
  23 #include <semaphore.h>
  24
  25 namespace Romain
  26 {
  27
  28 class RedundancyCallback;
  29
  30 #if WATCHDOG
  31 class Watchdog;
  32 #endif
  33
  34 struct Thread_group;
  35
  36 /*
  37  * Acting instance for a thread group
  38  *
  39  * A replicated thread does not possess a unique capability through which it is
  40  * accessible. However, Fiasco allows others to send a message to a thread
  41  * using its thread capability directly instead of an IPC gate. In order to
  42  * allow for this in the context of replicated thread groups, we replace the
  43  * original cap slot with an IPC gate that receives all these messages. A
  44  * dedicated thread, the GateAgent, is attached to this gate. It waits for
  45  * messages and delivers them to the replicas.
  46  *
  47  * Q: Does the GateAgent receive early / out-of-order messages?
  48  * A: No. The GateAgent only performs an operation if the replicated thread group
  49  *    decided to do so. It then only acts as their agent.
  50  *
  51  * Q: Do we need to distinguish external send and call operations?
  52  * A: No. A reply will only be sent if the thread group decides so. The gateagent will
  53  *    use the proper function required in every case.
  54  *
  55  * Q: IPC wait() does not specify a gate to wait on. How do we handle this?
  56  * A: We intercept all calls that bind the thread to a newly created gate. In
  57  *    every case, instead of binding a real thread, we bind the GateAgent to this
  58  *    gate. Thereby, an open wait is triggered by the thread group, but always
  59  *    carried out by the agent and therefore all messages arrive in the right
  60  *    place.
  61  *
  62  * Q: How do we handle IRQs / explicit receive gates?
  63  * A: The gate agent also handles those. To do so, it needs to be attached to these
  64  *    objects (as opposed to the original threads). Therefore, we need to intercept
  65  *    IPC gate creation (see the factory syscall handler) and IRQ object attachment
  66  *    (see syscall handling) and adapt their parameters to use the thread group's
  67  *    gate agent.
  68  */
  69
  70 #define USE_SHMSYNC 0
  71 #define USE_IRQ     1
  72
  73 struct GateAgent
  74 {
  75         enum { GateAgent_label = 0x195300 };
  76
  77         sem_t                 init_sem;        // semaphore for startup
  78         L4::Cap<L4::Kobject>  gate_cap;        // IPC gate the agent polls on
  79 #if USE_IRQ
  80         L4::Cap<L4::Irq>      gate_irq;        // IRQ to notify the agent of pending work
  81 #endif
  82         pthread_t             listener;        // gate agent thread
  83         L4::Cap<L4::Thread>   listener_cap;    // gate agent thread cap
  84
  85         Romain::Thread_group* owner_group;     // thread group this agent belongs to
  86         Romain::App_thread*   current_client;  // currently handled replica thread (will be
  87                                                // one of the group's threads)
  88
  89         static void *listener_function(void *gk);
  90
  91         /*
  92          * Trigger gate agent (replica -> agent)
  93          */
  94         void trigger_agent(Romain::App_thread *t)
  95         {
  96                 _check(current_client != 0, "current client not 0");
  97                 current_client = t;
  98 #if USE_SHMSYNC
  99                 while (current_client)
 100                         l4_thread_yield();
 101 #endif
 102
 103 #if USE_IRQ
 104                 gate_irq->trigger();
 105                 l4_umword_t lbl;
 106                 l4_ipc_wait(l4_utcb(), &lbl, L4_IPC_NEVER);
 107                 /* No error check. After return from this call, the UTCB contains
 108                  * the IPC error the gate agent received, which may in fact be a real
 109                  * IPC error -> it is up to the app to check and react on it. */
 110 #endif
 111
 112                 DEBUG() << "agent returned.";
 113         }
 114
 115
 116         GateAgent(l4_umword_t cap_idx, Romain::Thread_group *tg)
 117                 : owner_group(tg), current_client(0)
 118         {
 119                 DEBUG() << BOLD_RED << "GateAgent" << NOCOLOR;
 120                 sem_init(&init_sem, 0, 0);
 121
 122                 /*
 123                  * launch listener thread
 124                  */
 125                 l4_mword_t err  = pthread_create(&listener, 0,
 126                                                  listener_function, (void*)this);
 127                 _check(err != 0, "error creating listener thread");
 128                 l4_thread_yield();
 129
 130                 listener_cap    = L4::Cap<L4::Thread>(pthread_getl4cap(listener));
 131                 _check(!listener_cap.is_valid(), "could not get listener pthread cap?");
 132
 133                 /*
 134                  * create IPC gate
 135                  */
 136                 gate_cap        = L4::Cap<L4::Kobject>(cap_idx << L4_CAP_SHIFT);
 137                 _check(!gate_cap.is_valid(), "could not create gate");
 138                 l4_msgtag_t tag = L4Re::Env::env()->factory()->create_gate(gate_cap,
 139                                                                            listener_cap,
 140                                                                            GateAgent_label);
 141                 if (l4_error(tag)) {
 142                         enter_kdebug("gate creation error");
 143                 }
 144
 145 #if USE_IRQ
 146                 /*
 147                  * Create notification IRQ
 148                  *
 149                  * Q: Why no pthread semaphore or mutex?
 150                  * A: Evil things happened. Seems we disrupt some pthreads assumptions
 151                  *    when doing so in our context.
 152                  */
 153                 gate_irq = L4Re::Util::cap_alloc.alloc<L4::Irq>();
 154                 //_check(!gate_irq.valid(), "error allocating gate2");
 155                 tag = L4Re::Env::env()->factory()->create_irq(gate_irq);
 156                 if (l4_error(tag)) {
 157                         enter_kdebug("IRQ creation error");
 158                 }
 159                 tag = gate_irq->attach(GateAgent_label+2, listener_cap);
 160                 if (l4_error(tag)) {
 161                         enter_kdebug("attach error");
 162                 }
 163 #endif
 164
 165                 sem_post(&init_sem);
 166                 DEBUG() << "started agent thread";
 167         }
 168 };
 169
 170 /*
 171  * Instance of a replicated thread
 172  *
 173  * In contrast to an App_instance (which maps to a dedicated address space),
 174  * a thread group simply keeps track of which App_threads belong together
 175  * as they run the same original thread code.
 176  */
 177 struct Thread_group
 178 {
 179         Thread_group(std::string n, l4_umword_t cap_idx, l4_umword_t u)
 180                 : threads(), name(n), uid(u), stopped(true)
 181         {
 182                 DEBUG() << BOLD_RED << "Thread_group " << NOCOLOR << std::hex << cap_idx;
 183
 184                 gateagent = new GateAgent(cap_idx, this);
 185
 186                 sem_init(&activation_sem, 0, 0);
 187         }
 188
 189         /* Replica threads */
 190         std::vector<Romain::App_thread*> threads;
 191
 192         /* DBG: group name, unique ID */
 193         std::string name;
 194         l4_umword_t uid;
 195
 196         /* The group's gate agent. Docs, see there. */
 197         GateAgent* gateagent;
 198
 199         /*
 200          * Marks the thread group as currently stopped.
 201          *
 202          * - initially, we start in this state, because threads are created
 203          *   using factory_create and later run when activated through the
 204          *   scheduler
 205          * - later we track stopped states, e.g., when blocking in a system
 206          *   call
 207          */
 208         bool stopped;
 209
 210         /*
 211          * Barrier that is used to block vCPU until the thread gets
 212          * activated.
 213          */
 214         sem_t activation_sem;
 215
 216
 217         Romain::RedundancyCallback *redundancyCB;
 218         void set_redundancy_callback(Romain::RedundancyCallback* cb)
 219         { redundancyCB = cb; }
 220
 221 #if WATCHDOG
 222         Romain::Watchdog *watchdog;
 223         void set_watchdog(Romain::Watchdog *w)
 224         { watchdog = w; }
 225 #endif
 226
 227         void add_replica(App_thread *a)
 228         {
 229                 threads.push_back(a);
 230         }
 231
 232
 233         /* Halt all threads by setting their prio to 0 */
 234         void halt()
 235         {
 236                 INFO() << "Halting TG '" << name << "'";
 237                 for (auto it = threads.begin(); it != threads.end(); ++it) {
 238                         (*it)->halt();
 239                 }
 240         }
 241
 242
 243         /*
 244          * Activate the thread group
 245          *
 246          * notify all vCPUs blocking on the activation semaphore
 247          */
 248         void activate()
 249         {
 250                 stopped = false;
 251
 252                 for (l4_umword_t i = 0; i < threads.size(); ++i) {
 253                         sem_post(&activation_sem);
 254                 }
 255         }
 256
 257
 258         /*
 259          * Lets the current thread indicate being ready to run
 260          * by waiting on the activation semapore.
 261          */
 262         void ready()
 263         {
 264                 sem_wait(&activation_sem);
 265         }
 266
 267
 268         Romain::App_thread* get(l4_umword_t number)
 269         {
 270                 _check(number < threads.size(), "invalid thread instance requested");
 271                 return threads[number];
 272         }
 273
 274
 275         void ex_regs(Romain::App_thread *caller);
 276         void scheduler_run(Romain::App_thread* caller);
 277
 278         /*
 279          * Check if thread_control() was called with parameters
 280          * we don't support yet.
 281          */
 282         void sanity_check_control(l4_umword_t flags, l4_utcb_t *utcb);
 283         void control(Romain::App_thread *t, l4_utcb_t *utcb, Romain::App_model* am);
 284
 285         void gdt(Romain::App_thread *t, l4_utcb_t *utcb);
 286 };
 287 }