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() << BOLD_RED << "GateAgent" << NOCOLOR;
 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                 l4_thread_yield();
 125
 126                 listener_cap    = L4::Cap<L4::Thread>(pthread_getl4cap(listener));
 127                 _check(!listener_cap.is_valid(), "could not get listener pthread cap?");
 128
 129                 /*
 130                  * create IPC gate
 131                  */
 132                 gate_cap        = L4::Cap<L4::Kobject>(cap_idx << L4_CAP_SHIFT);
 133                 _check(!gate_cap.is_valid(), "could not create gate");
 134                 l4_msgtag_t tag = L4Re::Env::env()->factory()->create_gate(gate_cap,
 135                                                                            listener_cap,
 136                                                                            GateAgent_label);
 137                 if (l4_error(tag)) {
 138                         enter_kdebug("gate creation error");
 139                 }
 140
 141 #if USE_IRQ
 142                 /*
 143                  * Create notification IRQ
 144                  *
 145                  * Q: Why no pthread semaphore or mutex?
 146                  * A: Evil things happened. Seems we disrupt some pthreads assumptions
 147                  *    when doing so in our context.
 148                  */
 149                 gate_irq = L4Re::Util::cap_alloc.alloc<L4::Irq>();
 150                 //_check(!gate_irq.valid(), "error allocating gate2");
 151                 tag = L4Re::Env::env()->factory()->create_irq(gate_irq);
 152                 if (l4_error(tag)) {
 153                         enter_kdebug("IRQ creation error");
 154                 }
 155                 tag = gate_irq->attach(GateAgent_label+2, listener_cap);
 156                 if (l4_error(tag)) {
 157                         enter_kdebug("attach error");
 158                 }
 159 #endif
 160
 161                 sem_post(&init_sem);
 162                 DEBUG() << "started agent thread";
 163         }
 164 };
 165
 166 /*
 167  * Instance of a replicated thread
 168  *
 169  * In contrast to an App_instance (which maps to a dedicated address space),
 170  * a thread group simply keeps track of which App_threads belong together
 171  * as they run the same original thread code.
 172  */
 173 struct Thread_group
 174 {
 175         Thread_group(std::string n, unsigned cap_idx, unsigned u)
 176                 : threads(), name(n), uid(u), stopped(true)
 177         {
 178                 DEBUG() << BOLD_RED << "Thread_group " << NOCOLOR << std::hex << cap_idx;
 179
 180                 gateagent = new GateAgent(cap_idx, this);
 181
 182                 sem_init(&activation_sem, 0, 0);
 183         }
 184
 185         /* Replica threads */
 186         std::vector<Romain::App_thread*> threads;
 187
 188         /* DBG: group name, unique ID */
 189         std::string name;
 190         unsigned uid;
 191
 192         /* The group's gate agent. Docs, see there. */
 193         GateAgent* gateagent;
 194
 195         /*
 196          * Marks the thread group as currently stopped.
 197          *
 198          * - initially, we start in this state, because threads are created
 199          *   using factory_create and later run when activated through the
 200          *   scheduler
 201          * - later we track stopped states, e.g., when blocking in a system
 202          *   call
 203          */
 204         bool stopped;
 205
 206         /*
 207          * Barrier that is used to block vCPU until the thread gets
 208          * activated.
 209          */
 210         sem_t activation_sem;
 211
 212
 213         Romain::RedundancyCallback *redundancyCB;
 214         void set_redundancy_callback(Romain::RedundancyCallback* cb)
 215         { redundancyCB = cb; }
 216
 217
 218         void add_replica(App_thread *a)
 219         {
 220                 threads.push_back(a);
 221         }
 222
 223
 224         /* Halt all threads by setting their prio to 0 */
 225         void halt()
 226         {
 227                 INFO() << "Halting TG '" << name << "'";
 228                 for (auto it = threads.begin(); it != threads.end(); ++it) {
 229                         (*it)->halt();
 230                 }
 231         }
 232
 233
 234         /*
 235          * Activate the thread group
 236          *
 237          * notify all vCPUs blocking on the activation semaphore
 238          */
 239         void activate()
 240         {
 241                 stopped = false;
 242
 243                 for (unsigned i = 0; i < threads.size(); ++i) {
 244                         sem_post(&activation_sem);
 245                 }
 246         }
 247
 248
 249         /*
 250          * Lets the current thread indicate being ready to run
 251          * by waiting on the activation semapore.
 252          */
 253         void ready()
 254         {
 255                 sem_wait(&activation_sem);
 256         }
 257
 258
 259         Romain::App_thread* get(unsigned number)
 260         {
 261                 _check(number < threads.size(), "invalid thread instance requested");
 262                 return threads[number];
 263         }
 264
 265
 266         void ex_regs(Romain::App_thread *caller);
 267         void scheduler_run(Romain::App_thread* caller);
 268
 269         /*
 270          * Check if thread_control() was called with parameters
 271          * we don't support yet.
 272          */
 273         void sanity_check_control(unsigned flags, l4_utcb_t *utcb);
 274         void control(Romain::App_thread *t, l4_utcb_t *utcb, Romain::App_model* am);
 275
 276         void gdt(Romain::App_thread *t, l4_utcb_t *utcb);
 277 };
 278 }