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