shotgun/lib/environment.c

   1 #include "shotgun/lib/shotgun.h"
   2 #include "shotgun/lib/cpu.h"
   3 #include "shotgun/lib/environment.h"
   4 #include "shotgun/lib/config_hash.h"
   5 #include "shotgun/lib/symbol.h"
   6 #include "shotgun/lib/tuple.h"
   7
   8 #include <pthread.h>
   9 #include <signal.h>
  10 #include <ev.h>
  11 #include <unistd.h>
  12
  13 static pthread_key_t global_key;
  14
  15 #define lock(e) pthread_mutex_lock(&e->mutex)
  16 #define unlock(e) pthread_mutex_unlock(&e->mutex)
  17
  18 void environment_at_startup() {
  19   pthread_key_create(&global_key, NULL);
  20 }
  21
  22 static void _find_child_waiters(int key, void *mach) {
  23   machine m = mach;
  24   cpu_find_waiters(m->s);
  25 }
  26
  27
  28 static void _child_ev_cb(EV_P_ struct ev_signal *w, int revents) {
  29   environment e = (environment)w->data;
  30
  31   // call for all VMs/states
  32   ht_vconfig_each(e->machines, _find_child_waiters);
  33 }
  34
  35 environment environment_new() {
  36   environment e = ALLOC_N(struct rubinius_environment, 1);
  37   e->machines = ht_vconfig_create(11);
  38   e->machine_id = 1;
  39
  40   e->messages = ht_vconfig_create(11);
  41   e->sig_event_base = ev_default_loop(EVFLAG_FORKCHECK);
  42   return e;
  43 }
  44
  45 static void _fork_event_loops(int key, void *mach) {
  46   machine m = mach;
  47   ev_loop_fork(m->s->event_base);
  48   ev_loop(m->s->event_base, EVLOOP_NONBLOCK);
  49 }
  50
  51 void environment_fork() {
  52   environment e = environment_current();
  53
  54   ev_default_fork();
  55   ev_loop(e->sig_event_base, EVLOOP_NONBLOCK);
  56   ht_vconfig_each(e->machines, _fork_event_loops);
  57 }
  58
  59 void environment_setup_thread(environment e, machine m) {
  60   struct rubinius_global *global = ALLOC_N(struct rubinius_global, 1);
  61
  62   global->e = e;
  63   global->m = m;
  64
  65   pthread_setspecific(global_key, (const void*)global);
  66
  67   ev_signal_init(&e->sig_ev, _child_ev_cb, SIGCHLD);
  68   e->sig_ev.data = e;
  69   ev_signal_start(e->sig_event_base, &e->sig_ev);
  70 }
  71
  72 environment environment_current() {
  73   struct rubinius_global *global;
  74   global = (struct rubinius_global*)pthread_getspecific(global_key);
  75   return global->e;
  76 }
  77
  78 machine environment_current_machine() {
  79   struct rubinius_global *global;
  80   global = (struct rubinius_global*)pthread_getspecific(global_key);
  81   if(!global) return NULL;
  82   return global->m;
  83 }
  84
  85 void environment_add_machine(environment e, machine m) {
  86   int *key;
  87
  88   lock(e);
  89   m->pthread = pthread_self();
  90   m->id = e->machine_id++;
  91
  92   key = ALLOC_N(int, 1);
  93   *key = m->id;
  94
  95   ht_vconfig_insert(e->machines, key, (void*)m);
  96   unlock(e);
  97 }
  98
  99 int environment_del_machine(environment e, machine m) {
 100   machine old;
 101
 102   lock(e);
 103   old = ht_vconfig_remove(e->machines, &m->id);
 104   unlock(e);
 105
 106   assert(old == m);
 107   return TRUE;
 108 }
 109
 110 int environment_join_machine(environment e, int id) {
 111   machine m;
 112   void *ret;
 113
 114   lock(e);
 115   m = ht_vconfig_search(e->machines, &id);
 116   unlock(e);
 117
 118   if(!m) return FALSE;
 119
 120   if(pthread_join(m->pthread, &ret) == 0) {
 121     machine_destroy(m);
 122     return TRUE;
 123   }
 124
 125   return FALSE;
 126 }
 127
 128 void environment_exit_machine() {
 129   machine m = environment_current_machine();
 130   environment e = environment_current();
 131   STATE;
 132
 133   state = m->s;
 134
 135   if(m->parent_id) {
 136     environment_send_message(e, m->parent_id,
 137         tuple_new2(state, 2, SYM("machine_exited"), I2N(m->id)));
 138   }
 139
 140   pthread_exit(NULL);
 141 }
 142
 143 int environment_load_machine(environment e, machine m) {
 144   if(e->platform_config) {
 145     machine_parse_config_file(m, e->platform_config);
 146   }
 147
 148   machine_migrate_config(m);
 149
 150   if(m->s->excessive_tracing) {
 151     printf("[ Loading bootstrap bundle %s]\n", e->bootstrap_path);
 152   }
 153
 154   if(!machine_load_bundle(m, e->bootstrap_path)) {
 155     printf("Problem encountered while loading bootstrap %s\n", e->bootstrap_path);
 156     return FALSE;
 157   }
 158
 159   if(m->s->excessive_tracing) {
 160     printf("[ Loading platform bundle %s]\n", e->platform_path);
 161   }
 162
 163   if(!machine_load_bundle(m, e->platform_path)) {
 164     printf("Problem encountered while loading platform %s\n", e->platform_path);
 165     return FALSE;
 166   }
 167
 168   if(m->s->excessive_tracing) {
 169     printf("[ Loading core bundle %s]\n", e->core_path);
 170   }
 171
 172   if(!machine_load_bundle(m, e->core_path)) {
 173     printf("Problem encountered while loading core %s\n", e->core_path);
 174     return FALSE;
 175   }
 176
 177   if(!machine_run_file(m, e->loader_path)) {
 178     printf("Unclean exit from %s\n", e->core_path);
 179     return FALSE;
 180   }
 181
 182   return TRUE;
 183 }
 184
 185 /* NOTE: this duplicates rubinius_global from environment.h */
 186 struct thread_args {
 187   environment e;
 188   machine m;
 189 };
 190
 191 void *_environment_spawn(void *input) {
 192   sigset_t set;
 193   struct thread_args *args = (struct thread_args*)input;
 194
 195   environment_setup_thread(args->e, args->m);
 196
 197   /* For now, we mask all signals except VTALRM in sub machines. */
 198   sigfillset(&set);
 199   //sigdelset(&set, SIGVTALRM);
 200   pthread_sigmask(SIG_SETMASK, &set, NULL);
 201
 202   environment_load_machine(args->e, args->m);
 203   XFREE(args);
 204   return NULL;
 205 }
 206
 207 void environment_start_thread(environment e, machine m) {
 208   struct thread_args *args = ALLOC_N(struct thread_args, 1);
 209
 210   args->e = e;
 211   args->m = m;
 212
 213   pthread_create(&m->pthread, NULL, _environment_spawn, (void*)args);
 214 }
 215
 216 static const char magic[] = "!";
 217
 218 void environment_send_message(environment e, int id, OBJECT msg) {
 219   ptr_array cur;
 220   bstring data;
 221   machine m;
 222   int *key;
 223
 224   /* Marshal the data before the lock. */
 225   m = environment_current_machine();
 226   data = cpu_marshal_to_bstring(m->s, msg, 0);
 227
 228   lock(e);
 229
 230   cur = ht_vconfig_search(e->messages, &id);
 231   if(!cur) {
 232     cur = ptr_array_new(8);
 233     key = ALLOC_N(int, 1);
 234     *key = id;
 235     ht_vconfig_insert(e->messages, key, cur);
 236   }
 237
 238   ptr_array_append(cur, (xpointer)data);
 239
 240   m = ht_vconfig_search(e->machines, &id);
 241
 242   /* write the magic byte, to let the machine know there are
 243    * messages for it. */
 244   write(m->message_write_fd, magic, 1);
 245
 246   unlock(e);
 247 }
 248
 249 OBJECT environment_get_message(environment e, int id) {
 250   ptr_array cur;
 251   bstring data;
 252   machine m;
 253
 254   lock(e);
 255
 256   cur = ht_vconfig_search(e->messages, &id);
 257   if(!cur) goto error;
 258   if(ptr_array_length(cur) == 0) goto error;
 259   data = ptr_array_remove_index_ordered(cur, 0);
 260   if(!data) goto error;
 261
 262   /* Now that we're sure we've got what we need, we unlock.. */
 263   unlock(e);
 264
 265   /* and unmarshal. */
 266   m = environment_current_machine();
 267   return cpu_unmarshal(m->s, (uint8_t*)bdata(data), (int)blength(data), 0);
 268 error:
 269   unlock(e);
 270   return Qnil;
 271 }