Mercurial > repos > rhope
view worker.c @ 145:357f4ce3ca6d
Add incredibly ugly implementation of Read Delim to TCP Connection
author | Mike Pavone <pavone@retrodev.com> |
---|---|
date | Sun, 21 Nov 2010 22:08:17 -0500 |
parents | 914ad38f9b59 |
children |
line wrap: on
line source
#include "structs.h" #include "interp.h" #include "datum.h" #include "parser.h" #include "saveload.h" #include <stdlib.h> #include <string.h> int vis_worker_from_string(datum ** inputlist, queue_entry * worker_entry) { int i; worker_datum * work; datum * output = new_datum(BUILTIN_TYPE_WORKER, 1, sizeof(worker_datum), worker_entry->instance->def->program); work = output->c.generic.data; find_worker(inputlist[0]->c.generic.data, NULL, NULL, worker_entry->instance->def->program, &work->def); if(work->def) add_program_ref(work->def->program); for(i = 0; i < 32; ++i) work->params[i] = NULL; release_ref(inputlist[0]); inputlist[0] = output; return 0; } BOOL list_to_array(datum * list_dat, datum ** out_params, int num) { int i; datum * params[2]; list_data * list = list_dat->c.generic.data; if(num > list->num_entries) { return FALSE; } for(i = 0; i < num; ++i) out_params[i] = add_ref(list->entries[i]); release_ref(list_dat); return TRUE; } datum * array_to_list(datum ** in_params, int num, program * prog) { int i; datum * params[2]; params[0] = create_list(prog); for(i = 0; i < num; ++i) { params[1] = in_params[i]; vis_list_append(params, NULL); } return params[0]; } int worker_populate_inputs(datum * inputWorker, datum * ioList, datum ** inputlist) { int i,j=0; worker_datum * work = inputWorker->c.generic.data; list_data * list = ioList->c.generic.data; for(i = 0; i < work->def->num_inputs; ++i) if(work->params[i]) { DEBUGPRINTF("Setting input %d to %X from worker object\n", i, work->params[i]); inputlist[i] = add_ref(work->params[i]); } else { if(j < list->num_entries) { DEBUGPRINTF("Setting input %d to %X from input %d\n", i, list->entries[j], j); inputlist[i] = add_ref(list->entries[j]); ++j; } else { ERRORPRINTF("Error: List passed to Do@Worker doesn't have enough entries for worker %s\n", work->def->name); return -1; } } return 0; } int vis_worker_do(datum ** inputlist, queue_entry * worker_entry) { //int i,j; worker_datum * work; //worker_def * def; //list_data * list; datum * inputWorker = inputlist[0]; datum * ioList = inputlist[1]; int returnval; //release_ref(inputlist[0]); work = inputlist[0]->c.generic.data; //list_to_array(inputlist[1], inputlist, work->def->num_inputs); /*list = inputlist[1]->c.generic.data; j = 0; DEBUGPRINTF("vis_worker_do: def_name: %s, num_inputs = %d, list length = %d\n", work->def->name, work->def->num_inputs, list->num_entries); for(i = 0; i < work->def->num_inputs; ++i) if(work->params[i]) { DEBUGPRINTF("Setting input %d to %X from worker object\n", i, work->params[i]); inputlist[i] = add_ref(work->params[i]); } else { DEBUGPRINTF("Setting input %d to %X from input %d, company name: %s\n", i, list->entries[j], j, list->entries[j]->company->name); inputlist[i] = add_ref(list->entries[j]); ++j; }*/ returnval = worker_populate_inputs(inputWorker, ioList, inputlist); release_ref(ioList); if(returnval) { print_stack_trace(worker_entry->instance); return returnval; } if(work->def->program != worker_entry->instance->def->program) { prep_program(work->def->program ); } returnval = execute_def(work->def, *worker_entry, inputlist, pack_list_sub_callback); if(returnval == 0) { inputlist[0] = array_to_list(inputlist, work->def->num_outputs, worker_entry->instance->def->program); /*vis_list_new(params, NULL); DEBUGPRINTF("params[0] after vis_list_new: %X\n", params[0]); for(i = 0; i < work->def->num_outputs; ++i) { params[1] = inputlist[i]; vis_list_append(params, NULL); DEBUGPRINTF("params[0] after vis_list_append: %X\n", params[0]); } inputlist[0] = params[0];*/ } release_ref(inputWorker); return returnval; } int vis_worker_setinput(datum ** inputlist, queue_entry * worker_entry) { worker_datum * work; inputlist[0] = copy_datum(inputlist[0], 0); work = inputlist[0]->c.generic.data; if(inputlist[1]->c.integers.num_a < 32) work->params[inputlist[1]->c.integers.num_a] = inputlist[2]; else { puts("Error: Visuality currently only supports 32 inputs\n"); execute_active = FALSE; return -1; } release_ref(inputlist[1]); inputlist[1] = inputlist[2] = NULL; return 0; } int vis_worker_add_worker_call(datum ** inputlist, queue_entry * worker_entry) { worker_datum * add_work; worker_datum * work = inputlist[0]->c.generic.data; int index; add_work = inputlist[1]->c.generic.data; /*index = add_work->def - work->def->program->deflist; DEBUGPRINTF("work->def: %X, deflist: %X\n", work->def, worker_entry->instance->def->program->deflist); DEBUGPRINTF("work->def->name: %X\n", work->def->name);*/ index = add_worker_to_def(work->def, add_work->def, 1.0, 1.0); release_ref(inputlist[1]); inputlist[1] = new_datum(BUILTIN_TYPE_WHOLE, 2, 0, worker_entry->instance->def->program); inputlist[1]->c.integers.num_a = index; return 0; } int vis_worker_add_wire(datum ** inputlist, queue_entry * worker_entry) { worker_datum * work = inputlist[0]->c.generic.data; add_wire(work->def, inputlist[1]->c.integers.num_a, inputlist[2]->c.integers.num_a, inputlist[3]->c.integers.num_a, inputlist[4]->c.integers.num_a); release_ref(inputlist[1]); release_ref(inputlist[2]); release_ref(inputlist[3]); release_ref(inputlist[4]); return 0; } void list_change_program(list_data * list, program * new_prog) { int i; for(i = 0; i < list->num_entries; ++i) { if(list->entries[i] && list->entries[i]->company->type_id < USER_DEFINED_TYPES) { list->entries[i] = copy_datum(list->entries[i], 0); list->entries[i]->company = new_prog->companylist + list->entries[i]->company->type_id; if(list->entries[i]->company->type_id == BUILTIN_TYPE_LIST) list_change_program(list->entries[i]->c.generic.data, new_prog); } } } int vis_worker_add_constant(datum ** inputlist, queue_entry * worker_entry) { datum * constant = inputlist[1]; worker_datum * work = inputlist[0]->c.generic.data; int index = add_constant(work->def, "code generated", 1.0, 1.0); if(worker_entry->instance->def->program != work->def->program && constant->company->type_id < USER_DEFINED_TYPES) { constant = copy_datum(constant, 0); constant->company = work->def->program->companylist + constant->company->type_id; if(constant->company->type_id == BUILTIN_TYPE_LIST) { list_change_program(constant->c.generic.data, work->def->program); } } work->def->implement_func->workerlist[index].value_index = constant; inputlist[1] = new_datum(BUILTIN_TYPE_WHOLE, 2, 0, worker_entry->instance->def->program); inputlist[1]->c.integers.num_a = index; return 0; } int vis_worker_add_input(datum ** inputlist, queue_entry * worker_entry) { worker_datum * work = inputlist[0]->c.generic.data; int index = add_input_num(work->def, inputlist[1]->c.generic.data, inputlist[2]->c.integers.num_a, 1.0, 1.0); release_ref(inputlist[1]); release_ref(inputlist[2]); inputlist[1] = new_datum(BUILTIN_TYPE_WHOLE, 2, 0, worker_entry->instance->def->program); inputlist[1]->c.integers.num_a = index; return 0; } int vis_worker_add_output(datum ** inputlist, queue_entry * worker_entry) { worker_datum * work = inputlist[0]->c.generic.data; int index = add_output_num(work->def, inputlist[1]->c.generic.data, inputlist[2]->c.integers.num_a, 1.0, 1.0); release_ref(inputlist[1]); release_ref(inputlist[2]); inputlist[1] = new_datum(BUILTIN_TYPE_WHOLE, 2, 0, worker_entry->instance->def->program); inputlist[1]->c.integers.num_a = index; return 0; } int vis_worker_add_objectget(datum ** inputlist, queue_entry * worker_entry) { worker_datum * work = inputlist[0]->c.generic.data; int index = add_get_comp_room(work->def, inputlist[1]->c.generic.data, 1.0,1.0); release_ref(inputlist[1]); inputlist[1] = new_datum(BUILTIN_TYPE_WHOLE, 2, 0, worker_entry->instance->def->program); inputlist[1]->c.integers.num_a = index; return 0; } int vis_worker_add_objectset(datum ** inputlist, queue_entry * worker_entry) { worker_datum * work = inputlist[0]->c.generic.data; int index = add_set_comp_room(work->def, inputlist[1]->c.generic.data, 1.0,1.0); release_ref(inputlist[1]); inputlist[1] = new_datum(BUILTIN_TYPE_WHOLE, 2, 0, worker_entry->instance->def->program); inputlist[1]->c.integers.num_a = index; return 0; } int vis_worker_add_globalget(datum ** inputlist, queue_entry * worker_entry) { worker_datum * work = inputlist[0]->c.generic.data; int index; char * name = malloc(inputlist[1]->c.generic.len + inputlist[2]->c.generic.len + 1); memcpy(name, inputlist[1]->c.generic.data, inputlist[1]->c.generic.len); release_ref(inputlist[1]); strcat(name, "::"); strcat(name, inputlist[2]->c.generic.data); release_ref(inputlist[2]); index = add_global_get(work->def, name, 1.0,1.0); free(name); inputlist[1] = new_datum(BUILTIN_TYPE_WHOLE, 2, 0, worker_entry->instance->def->program); inputlist[1]->c.integers.num_a = index; return 0; } int vis_worker_add_globalset(datum ** inputlist, queue_entry * worker_entry) { worker_datum * work = inputlist[0]->c.generic.data; int index; char * name = malloc(inputlist[1]->c.generic.len + inputlist[2]->c.generic.len + 1); memcpy(name, inputlist[1]->c.generic.data, inputlist[1]->c.generic.len); release_ref(inputlist[1]); strcat(name, "::"); strcat(name, inputlist[2]->c.generic.data); release_ref(inputlist[2]); index = add_global_set(work->def, name, 1.0,1.0); free(name); inputlist[1] = new_datum(BUILTIN_TYPE_WHOLE, 2, 0, worker_entry->instance->def->program); inputlist[1]->c.integers.num_a = index; return 0; } int vis_worker_new(datum ** inputlist, queue_entry * worker_entry) { int i; datum * output = new_datum(BUILTIN_TYPE_WORKER, 1, sizeof(worker_datum), worker_entry->instance->def->program); worker_datum * work = output->c.generic.data; work->def = create_worker(worker_entry->instance->def->program, inputlist[0]->c.generic.data, 0, 0, USER_FLAG | WORKER_TYPE); add_program_ref(work->def->program); for(i = 0; i < 32; ++i) work->params[i] = NULL; release_ref(inputlist[0]); inputlist[0] = output; return 0; } int vis_worker_clear(datum ** inputlist, queue_entry * worker_entry) { int i; worker_datum * work = inputlist[0]->c.generic.data; VIS_EnterCriticalSection(work->def->implement_func->lock); for(i = 0; i < work->def->implement_func->num_workers; ++i) { if(work->def->implement_func->workerlist[i].type == CONSTANT) { release_ref(work->def->implement_func->workerlist[i].value_index); } } work->def->num_inputs = 0; work->def->num_outputs = 0; work->def->implement_func->num_workers = 0; work->def->implement_func->num_wires = 0; work->def->implement_func->dirty = TRUE; VIS_LeaveCriticalSection(work->def->implement_func->lock); return 0; } int vis_worker_uses(datum ** inputlist, queue_entry * worker_entry) { int i; worker_datum * work = inputlist[0]->c.generic.data; list_data * list = inputlist[1]->c.generic.data; char ** new_uses; if(list->num_entries) { new_uses = malloc(sizeof(char *) * list->num_entries); for(i = 0; i < list->num_entries; ++i) { if(list->entries[i]->company->type_id != BUILTIN_TYPE_STRING) { ERRORPUTS("List passed to Uses@Worker contains non-string value(s)\n"); execute_active = FALSE; return -1; } new_uses[i] = malloc(sizeof(char) * list->entries[i]->c.generic.len); memcpy(new_uses[i], list->entries[i]->c.generic.data, list->entries[i]->c.generic.len); } } else new_uses = NULL; if(work->def->uses_stores) free(work->def->uses_stores); work->def->uses_stores = new_uses; work->def->num_stores = list->num_entries; release_ref(inputlist[1]); return 0; } int vis_worker_setio_counts(datum ** inputlist, queue_entry * worker_entry) { int i; int new_inputs = inputlist[1]->c.integers.num_a; worker_datum * work = inputlist[0]->c.generic.data; release_ref(inputlist[1]); if(work->def->num_inputs < new_inputs) { work->def->input_types = realloc(work->def->input_types, sizeof(short)*new_inputs); for(i = work->def->num_inputs; i < new_inputs; ++i) work->def->input_types[i] = ANY_TYPE; } work->def->num_inputs = new_inputs; work->def->num_outputs = inputlist[2]->c.integers.num_a; release_ref(inputlist[2]); return 0; } int vis_program_new(datum ** inputlist, queue_entry * worker_entry) { program * prog = new_program(START_DEF_STORAGE, START_COMP_STORAGE); inputlist[0] = new_datum(BUILTIN_TYPE_PROGRAM, 2, 0, worker_entry->instance->def->program); inputlist[0]->c.generic.data = prog; inputlist[0]->c.generic.len = sizeof(program); //inputlist[0]->union_type = 1; return 0; } int vis_program_new_worker(datum ** inputlist, queue_entry * worker_entry) { int i; datum * output = new_datum(BUILTIN_TYPE_WORKER, 1, sizeof(worker_datum), worker_entry->instance->def->program); worker_datum * work = output->c.generic.data; add_program_ref(inputlist[0]->c.generic.data); work->def = create_worker(inputlist[0]->c.generic.data, inputlist[1]->c.generic.data, 0, 0, USER_FLAG | WORKER_TYPE); for(i = 0; i < 32; ++i) work->params[i] = NULL; release_ref(inputlist[1]); inputlist[1] = output; return 0; } int vis_program_add_worker(datum ** inputlist, queue_entry * worker_entry) { int i; worker_datum * work = inputlist[1]->c.generic.data; program * prog = inputlist[0]->c.generic.data; datum * output = new_datum(BUILTIN_TYPE_WORKER, 1, sizeof(worker_datum), worker_entry->instance->def->program); worker_datum * out_work = output->c.generic.data; add_program_ref(inputlist[0]->c.generic.data); out_work->def = create_worker(prog, work->def->name, work->def->num_inputs, work->def->num_outputs, WORKER_TYPE); out_work->def->implement_func = work->def->implement_func; out_work->def->type = work->def->type; if(work->def->type & USER_FLAG) out_work->def->program = work->def->program; for(i = 0; i < work->def->num_inputs; ++i) out_work->def->input_types[i] = out_work->def->input_types[i]; release_ref(inputlist[1]); inputlist[1] = output; return 0; } int vis_program_add_builtins(datum ** inputlist, queue_entry * worker_entry) { initpredefworkers((program *)inputlist[0]->c.generic.data); return 0; } int vis_program_run(datum ** inputlist, queue_entry * worker_entry) { datum * params[32]; defchunk * current; int i; program * prog = inputlist[0]->c.generic.data; params[0] = inputlist[1]; VIS_EnterCriticalSection(program_count_lock); ++program_count; VIS_LeaveCriticalSection(program_count_lock); prep_program(prog); add_program_ref(prog); init_custom_worker(-1, NULL, prog->defs->deflist, main_callback, add_ref(inputlist[0]), params); return 0; } int vis_program_find_worker(datum ** inputlist, queue_entry * worker_entry) { int i,returnval; worker_datum * work; worker_def * def; program * prog = inputlist[0]->c.generic.data; returnval = find_worker(inputlist[1]->c.generic.data, NULL, NULL, prog, &def); release_ref(inputlist[0]); if(returnval >= 0) { add_program_ref(prog); release_ref(inputlist[1]); inputlist[0] = new_datum(BUILTIN_TYPE_WORKER, 1, sizeof(worker_datum), worker_entry->instance->def->program); inputlist[1] = NULL; work = inputlist[0]->c.generic.data; work->def = def; for(i = 0; i < 32; ++i) work->params[i] = NULL; } else { inputlist[0] = NULL; } return 0; } void free_worker(worker_def * def) { int i; if(def->type & USER_FLAG && (def->type & TYPE_MASK) == WORKER_TYPE) { for(i = 0; i < def->implement_func->num_workers; ++i) if(def->implement_func->workerlist[i].type == CONSTANT) release_ref(def->implement_func->workerlist[i].value_index); free(def->implement_func->workerlist); free(def->implement_func->wirelist); free(def->implement_func->workers_to_wires_up); free(def->implement_func->workers_to_wires_down); free(def->implement_func); } free(def->name); free(def->input_types); free(def->output_types); } void free_company(company * comp) { free(comp->methodlist); free(comp->room_list); } void add_program_ref(program * prog) { VIS_EnterCriticalSection(prog->lock); ++prog->refcount; VIS_LeaveCriticalSection(prog->lock); } void release_program_ref(program * prog) { int i; defchunk * current, *temp; VIS_EnterCriticalSection(prog->lock); --prog->refcount; if(!prog->refcount) { VIS_LeaveCriticalSection(prog->lock); VIS_DeleteCriticalSection(prog->lock); current = prog->defs; while(current) { for(i = 0; i < current->num_defs; ++i) { free_worker(current->deflist + i); } current = current->next; } current = prog->defs; while(current) { temp = current; current = current->next; free(temp); } //FIXME: We can't currently free the object definitions because of issues related to global stores //for(i = 0; i < prog->num_companies; ++i) // free_company(prog->companylist + i); //free(prog->companylist); free(prog); } else { VIS_LeaveCriticalSection(prog->lock); } } /* int vis_worker_begin_transaction(datum ** inputlist, queue_entry * worker_entry) { worker_def * def; int returnval; datum * params[2]; int i; global_store * store; list_data * list = inputist[0]->c.generic.data; transaction * trans = malloc(sizeof(transaction) + list->num_entries - 1); trans->num_stores = list->num_entries; VIS_EnterCriticalSection(global_store_lock); for(i = 0; i < list->num_entries; ++i) { params[0] = global_store_dict; params[1] = add_ref(list->entries[i]); vis_dict_index(params, NULL); store = params[0]->c.generic.data; while(store->inuse) { VIS_LeaveCriticalSection(global_store_lock); #ifdef WIN32 Sleep(0); #else sleep(0); #endif VIS_EnterCriticalSection(global_store_lock); } store->inuse; trans->stores[i] = params[0]; } VIS_LeaveCriticalSection(global_store_lock; release_ref(inputlist[0]); def = deflist+inputlist[1]->c.integers.num_a; release_ref(inputlist[1]); list_to_array(inputlist[2], inputlist, def->num_inputs); returnval = execute_def_data(def, *worker_entry, inputlist, trasnaction_sub_callback, transaction); if(returnval == 0) inputlist[0] = array_to_list(inputlist, def->num_outputs); return returnval; } */