Skip to content

Commit

Permalink
Fixed or worked-around concurrency bugs. Implemented context join. No…
Browse files Browse the repository at this point in the history 
…w everything essentially works, except context destruction.
  • Loading branch information
@lucasaiu
lucasaiu committed Jan 11, 2013
1 parent 2a2a02f commit 96ea355
Show file tree
Hide file tree
Showing 8 changed files with 134 additions and 125 deletions.
Binary file modified boot/ocamlrun
View file
Binary file not shown.
2 changes: 1 addition & 1 deletion byterun/compact.c
View file
Expand Up @@ -115,7 +115,7 @@ static void invert_root_r (CAML_R, value v, value *p)
invert_pointer_at_r (ctx, (word *) p);
}

static char *compact_fl;
//static char *compact_fl;

static void init_compact_allocate_r (CAML_R)
{
Expand Down
65 changes: 27 additions & 38 deletions byterun/context.c
View file
Expand Up @@ -327,6 +327,9 @@ section. */
ctx->caml_stat_heap_chunks = 0;
/* ctx->caml_percent_max */

/* from compact.c */
/* ctx->compact_fl */

/* from callback.c */
ctx->caml_callback_depth = 0;
ctx->callback_code_threaded = 0;
Expand Down Expand Up @@ -448,7 +451,7 @@ void caml_scan_caml_globals_r(CAML_R, scanning_action f){

if(caml_get_thread_local_context()->descriptor->kind == caml_global_context_nonmain_local){
if(caml_global_no != 0)
{fprintf(stderr, "Context %p: scanning the %i Caml globals\n", ctx, caml_global_no); fflush(stderr);}
{/* fprintf(stderr, "Context %p: scanning the %i Caml globals\n", ctx, caml_global_no); fflush(stderr); */}
else
{fprintf(stderr, "~~~~~~~~~~~~~~~~~~~~~~~ Context %p: there are no Caml globals to scan [!!!]\n", ctx); fflush(stderr);}
}
Expand Down Expand Up @@ -547,40 +550,22 @@ library_context *caml_get_library_context_r(
return uctx;
}

// FIXME: caml_realloc_global_r (meta.c) is a dummy stub on the native compiler. --Luca Saiu REENTRANTRUNTIME

/* void caml_resize_global_array_r(CAML_R, size_t requested_global_no){ */
/* size_t old_global_no = Wosize_val(ctx->caml_global_data); */
/* size_t new_global_no = old_global_no; */
/* value new_array; */
/* int i; */

/* fprintf(stderr, "** old size: %i\n", (int)old_global_no); */
/* fprintf(stderr, "** requested new size: %i\n", (int)requested_global_no); */
/* /\* Don't ever shrink: *\/ */
/* if(requested_global_no <= old_global_no) */
/* return; */

/* /\* Find a size large enough to accommodate the given global no, but potentially larger. */
/* We want to minimize the number of resizing: *\/ */
/* while(new_global_no < requested_global_no) */
/* new_global_no *= 2; */

/* /\* Copy the already-used elements, and simply zero the rest: *\/ */
/* new_array = caml_alloc_shr_r(ctx, new_global_no, 0); */
/* fprintf(stderr, "** caml_resize_global_array_r: we have now reserved space for %i globals\n", (int)new_global_no); */
/* for (i = 0; i < old_global_no; i ++) */
/* caml_initialize_r(ctx, &Field(new_array, i), Field(ctx->caml_global_data, i)); */
/* for (; i < new_global_no; i ++) */
/* caml_initialize_r(ctx, &Field(new_array, i), Val_long(0)); */

/* /\* Make the new global array the "official" one for this context, */
/* and oldify it so we don't waste time scanning it too many times. */
/* The old global array will be garbage-collected. *\/ */
/* ctx->caml_global_data = new_array; */
/* caml_oldify_one_r (ctx, ctx->caml_global_data, &ctx->caml_global_data); */
/* caml_oldify_mopup_r (ctx); */
/* } */
extern void caml_destroy_context(CAML_R){
return; /////////////////////////////////////////////////////////
fprintf(stderr, "caml_destroy_context [context %p] [thread %p]: FIXME: really do it\n", ctx, (void*)(pthread_self())); fflush(stderr);

/* No global variables are live any more; destroy everything in the Caml heap: */
caml_shrink_extensible_buffer(&ctx->caml_globals, ctx->caml_globals.used_size);
//caml_gc_compaction_r(ctx, Val_unit); //!!!!!@@@@@@@@@@@@@@??????????????????
caml_stat_free(ctx->caml_globals.array);

// Free every dynamically-allocated object which is pointed by the context data structure [FIXME: really do it]:
caml_stat_free(ctx->descriptor);
fprintf(stderr, "caml_destroy_context [context %p] [thread %p]: FIXME: actually free everything\n", ctx, (void*)(pthread_self())); fflush(stderr);

/* Free the context data structure ifself: */
caml_stat_free(ctx);
}

/* The index of the first word in caml_globals which is not used yet.
This variable is shared by all contexts, and accessed in mutual
Expand All @@ -594,9 +579,10 @@ void caml_register_module_r(CAML_R, size_t size_in_bytes, long *offset_pointer){
char *module_name = (char*)offset_pointer + sizeof(long);

Assert(size_in_words * sizeof(void*) == size_in_bytes); /* there's a whole number of globals */
fprintf(stderr, "Context %p: ??????? caml_register_module_r [%s]: BEGIN [%lu bytes at %p]\n",
fprintf(stderr, "caml_register_module_r [context %p]: registering %s%p [%lu bytes at %p]: BEGIN\n",
ctx,
module_name,
offset_pointer,
(unsigned long)size_in_bytes,
offset_pointer); fflush(stderr);

Expand All @@ -620,10 +606,13 @@ void caml_register_module_r(CAML_R, size_t size_in_bytes, long *offset_pointer){
/* fprintf(stderr, "The offset (in bytes) we just wrote at %p is %li\n", offset_pointer, *offset_pointer); */
/* fprintf(stderr, "The context is at %p\n", (void*)ctx); */
/* fprintf(stderr, "Globals are at %p\n", (void*)ctx->caml_globals.array); */
fprintf(stderr, "caml_register_module_r: registered %p [%s]. END (still alive)\n", offset_pointer, module_name); fflush(stderr);
fprintf(stderr, "caml_register_module_r [context %p]: registered %s@%p. END (still alive)\n", ctx, module_name, offset_pointer); fflush(stderr);
}

void caml_after_module_initialization_r(CAML_R, size_t size_in_bytes, long *offset_pointer){
/* We keep the module name right after the offset pointer, as a read-only string: */
char *module_name = (char*)offset_pointer + sizeof(long);
fprintf(stderr, "caml_after_module_initialization_r [context %p]: %s@%p: still alive.\n", ctx, module_name, offset_pointer); fflush(stderr);
/*
fprintf(stderr, "caml_after_module_initialization_r: BEGIN [%lu bytes at %p]\n",
(unsigned long)size_in_bytes,
Expand Down Expand Up @@ -694,7 +683,7 @@ void caml_context_initialize_global_stuff(void){
exit(EXIT_FAILURE);
}
pthread_mutex_init(&caml_global_mutex, &attributes);
fprintf(stderr, "= {%u %p | %p}\n", caml_global_mutex.__data.__count, (void*)(long)caml_global_mutex.__data.__count, (void*)(pthread_self())); fflush(stderr);
//fprintf(stderr, "= {%u %p | %p}\n", caml_global_mutex.__data.__count, (void*)(long)caml_global_mutex.__data.__count, (void*)(pthread_self())); fflush(stderr);
pthread_mutexattr_destroy(&attributes);
}

Expand Down
8 changes: 8 additions & 0 deletions byterun/context.h
View file
Expand Up @@ -461,6 +461,8 @@ struct caml_global_context {
intnat caml_stat_heap_chunks; /* = 0; */
uintnat caml_percent_max; /* used in gc_ctrl.c and memory.c */

/* from compact.c */
char *compact_fl;

/* from callback.c */
int caml_callback_depth; /* = 0; */
Expand Down Expand Up @@ -577,6 +579,8 @@ struct caml_global_context_descriptor* caml_global_context_descriptor_of_value(v
#define INIT_CAML_R CAML_R = caml_get_thread_local_context()

extern caml_global_context *caml_initialize_first_global_context(void);
extern caml_global_context *caml_make_empty_context(void); /* defined in startup.c */
extern void caml_destroy_context(caml_global_context *c);

/* Access a thread-local context pointer */
extern caml_global_context *caml_get_thread_local_context(void);
Expand Down Expand Up @@ -807,6 +811,10 @@ extern library_context *caml_get_library_context_r(

#endif

#ifdef CAML_CONTEXT_COMPACT
#define compact_fl ctx->compact_fl
#endif

#ifdef CAML_CONTEXT_GC_CTRL
#ifndef NATIVE_CODE
#define caml_max_stack_size ctx->caml_max_stack_size
Expand Down
69 changes: 47 additions & 22 deletions byterun/context_split.c
View file
Expand Up @@ -183,6 +183,7 @@ static void caml_install_globals_and_data_as_c_byte_array_r(CAML_R, value *to_va
/* Deserialize globals and data from the byte array, and access each
element of the pair. */
//fprintf(stderr, "Context %p: L0 [thread %p]\n", ctx, (void*)(pthread_self())); fflush(stderr);
caml_acquire_global_lock();
globals_and_data =
caml_input_value_from_block_r(ctx,
globals_and_data_as_c_array,
Expand All @@ -191,6 +192,7 @@ static void caml_install_globals_and_data_as_c_byte_array_r(CAML_R, value *to_va
I don't want to mess up the interface myself, since I'm doing a lot of other
invasive changes --Luca Saiu REENTRANTRUNTIME */
LONG_MAX);
caml_release_global_lock();
//fprintf(stderr, "Context %p: L1 [thread %p]\n", ctx, (void*)(pthread_self())); fflush(stderr);
//caml_input_value_from_malloc_r(ctx, globals_and_data_as_c_array, 0); // this also frees the buffer */
global_tuple = Field(globals_and_data, 0);
Expand Down Expand Up @@ -224,17 +226,28 @@ static char* caml_serialize_context(CAML_R, value function)
CAMLreturnT(char*, result);
}

static void caml_deserialize_and_run_in_this_thread(char *blob, int index, sem_t *semaphore, /*out*/caml_global_context **to_context)
/* Return 0 on success and non-zero on failure. */
static int caml_run_in_this_thread_r(CAML_R, value function, int index)
{
CAML_R = caml_make_empty_context(); // this also sets the thread-local context
// FIXME: move part of caml_deserialize_and_run_in_this_thread here.
}

/* Return 0 on success and non-zero on failure. */
static int caml_deserialize_and_run_in_this_thread(char *blob, int index, sem_t *semaphore, /*out*/caml_global_context **to_context)
{
int did_we_fail;
CAML_R = caml_make_empty_context(); // ctx also becomes the thread-local context
CAMLparam0();
CAMLlocal2(function, result_or_exception);
*to_context = ctx;

fprintf(stderr, "======Forcing a GC\n"); fflush(stderr);
caml_gc_compaction_r(ctx, Val_unit); //!!!!!
fprintf(stderr, "======It's ok to have warnings about the lack of globals up to this point\n"); fflush(stderr);

//fprintf(stderr, "W0[context %p] [thread %p] (index %i) BBBBBBBBBBBBBBBBBBBBBBBBBB\n", ctx, (void*)(pthread_self()), index); fflush(stderr); caml_acquire_global_lock(); // FIXME: a test. this is obviously unusable in production
fprintf(stderr, "W1 [context %p] ctx->caml_local_roots is %p\n", ctx, caml_local_roots); fflush(stderr);
/* Make a new context, and deserialize the blob into it: */
fprintf(stderr, "W2 [context %p] [thread %p] (index %i)\n", ctx, (void*)(pthread_self()), index); fflush(stderr);
caml_gc_compaction_r(ctx, Val_unit); //!!!!!
fprintf(stderr, "W3 [context %p] [thread %p] (index %i) (function %p)\n", ctx, (void*)(pthread_self()), index, (void*)function); fflush(stderr);

// Allocate some trash:
Expand All @@ -251,8 +264,7 @@ static void caml_deserialize_and_run_in_this_thread(char *blob, int index, sem_t
/* caml_compact_heap_r (ctx); */
/* caml_final_do_calls_r (ctx); */

fprintf(stderr, "W5 [context %p] [thread %p] (index %i) (function %p)\n", ctx, (void*)(pthread_self()), index, (void*)function); fflush(stderr);
*to_context = ctx;
fprintf(stderr, "W5 [context %p] [thread %p] (index %i) (function %p). About to V the semaphore.\n", ctx, (void*)(pthread_self()), index, (void*)function); fflush(stderr);

/* We're done with the blob: unpin it via the semaphore, so that it
can be destroyed when all threads have deserialized. */
Expand All @@ -265,20 +277,29 @@ static void caml_deserialize_and_run_in_this_thread(char *blob, int index, sem_t
fprintf(stderr, "W7 [context %p] [thread %p] (index %i) (%i globals) ctx->caml_local_roots is %p\n", ctx, (void*)(pthread_self()), index, (int)(ctx->caml_globals.used_size / sizeof(value)), caml_local_roots); fflush(stderr);
caml_dump_global_mutex();

/* It's important that Extract_exception be used before the next
collection, because result_or_exception is an invalid value in
case of exception: */
result_or_exception = caml_callback_exn_r(ctx, function, Val_int(index));
if(Is_exception_result(result_or_exception)){
fprintf(stderr, "W7.5 [context %p] [thread %p] (index %i): the ocaml code raised an exception: FIXME: implement this case\n", ctx, (void*)(pthread_self()), index); fflush(stderr);
exit(EXIT_FAILURE);
} // if
//int i;for(i=0;i<3;i++){
did_we_fail = Is_exception_result(result_or_exception);
if(did_we_fail){
/* FIXME: we can't just do "caml_raise_r(ctx, Extract_exception(result_or_exception));".
If we want to propagate the exception to the parent context we
have to serialize the exception object, and then deserialize it
and raise it in the parent context. Is that useful? */
result_or_exception = Extract_exception(result_or_exception);
/* FIXME: shall we do something with the result? Really? It's simpler to just discard it. */
}
//}
/* Ok, we're done with ctx. Free its resources, and we're done: */

fprintf(stderr, "W8 [context %p] [thread %p] (index %i): STILL ALIVE AFTER RUNNING THE OCAML CODE\n", ctx, (void*)(pthread_self()), index); fflush(stderr);
#ifdef NATIVE_CODE
//fprintf(stderr, "@@@@@ In the child context caml_bottom_of_stack is %p\n", caml_bottom_of_stack);
#endif // #ifdef NATIVE_CODE
caml_gc_compaction_r(ctx, Val_unit); //!!!!!@@@@@@@@@@@@@@
fprintf(stderr, "W9 [context %p] [thread %p] (index %i)\n", ctx, (void*)(pthread_self()), index); fflush(stderr);
//caml_dump_global_mutex();
CAMLreturn0;
/* FIXME: divide this functions into two parts, so that we can call
caml_destroy_context out of a CAMLparamX...CAMLreturnX block. */
caml_destroy_context(ctx);

CAMLreturnT(int, did_we_fail);
}

struct caml_thread_arguments{
Expand All @@ -294,11 +315,11 @@ static void* caml_deserialize_and_run_in_this_thread_as_thread_function(void *ar
//fprintf(stderr, "Q0 (index %i)\n", args->index);
//sleep(12); // FIXME: !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
//fprintf(stderr, "Q1 (index %i)\n", args->index);
caml_deserialize_and_run_in_this_thread(args->blob, args->index, args->semaphore, args->split_contexts + args->index);
fprintf(stderr, "Q2 (index %i) [about to free args]\n", args->index); fflush(stderr);
int did_we_fail = caml_deserialize_and_run_in_this_thread(args->blob, args->index, args->semaphore, args->split_contexts + args->index);
fprintf(stderr, "Q2 (index %i) [about to free args]. Did we fail? %i\n", args->index, did_we_fail); fflush(stderr);
caml_stat_free(args);
fprintf(stderr, "Q3 (index %i): about to exit the thread\n", args->index); fflush(stderr);
return NULL;
return (void*)(long)did_we_fail;
}
static void caml_split_and_destroy_blob_r(CAML_R, char *blob, caml_global_context **split_contexts, size_t how_many, sem_t *semaphore)
{
Expand Down Expand Up @@ -385,6 +406,8 @@ CAMLprim value caml_context_split_r(CAML_R, value function, value thread_no_as_v
CAMLprim value caml_context_join_r(CAML_R, value context_as_value){
struct caml_global_context_descriptor *descriptor;
int pthread_join_result;
void* did_we_fail_as_void_star;
int did_we_fail;
CAMLparam1(context_as_value);
CAMLlocal1(result);
descriptor = caml_global_context_descriptor_of_value(context_as_value);
Expand All @@ -397,7 +420,9 @@ CAMLprim value caml_context_join_r(CAML_R, value context_as_value){
caml_failwith_r(ctx, "caml_context_join_r: remote context");
Assert(descriptor->kind == caml_global_context_nonmain_local);
//fprintf(stderr, "!!!! JOINING %p\n", (void*)descriptor->content.local_context.context->thread); fflush(stderr);
pthread_join_result = pthread_join(descriptor->content.local_context.context->thread, NULL);
pthread_join_result = pthread_join(descriptor->content.local_context.context->thread, &did_we_fail_as_void_star);
did_we_fail = (int)(long)did_we_fail_as_void_star;
fprintf(stderr, "!!!! JOINED %p: did we fail? %i\n", (void*)descriptor->content.local_context.context->thread, did_we_fail); fflush(stderr);
if(pthread_join_result != 0)
caml_failwith_r(ctx, "caml_context_join_r: pthread_join failed");
CAMLreturn(Val_unit);
Expand Down
2 changes: 1 addition & 1 deletion byterun/gc_ctrl.c
View file
Expand Up @@ -452,7 +452,7 @@ CAMLprim value caml_gc_compaction_r(CAML_R, value v)
//caml_finish_major_cycle_r (ctx);
//caml_compact_heap_r (ctx);
//caml_final_do_calls_r (ctx);
return Val_unit;
//return Val_unit;
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! EXPERIMENTAL: END
caml_gc_message (0x10, "Heap compaction requested\n", 0);
caml_empty_minor_heap_r (ctx);
Expand Down

0 comments on commit 96ea355

Please sign in to comment.