static int
do_execute(int node, const char *app)
{
	char *code;
	char *res;
	size_t code_size;
	size_t res_size;
	int e;

	log_debug(LOG_JOB, "Execute(%s,%s)\n", model_get_path(node), app);

	bk_app = (char *) app;

	csl_setup();

	if (0 != db_get_code(model_parent(node), app, &code, &code_size)) return -1;
	e = csl_execute(code, code_size, model_get_method(node), &res, &res_size);
	if (e) {
		send_result(CMD_ERROR, "err", 3);
	} else {
		send_result(CMD_RESULT, res, res_size);
	}
	free(res);

	csl_cleanup();

	return e;
}

static int
do_register(int node, const char *app, const char *fname)
{
	char *buf;
	char *code;
	size_t codelen;
	int e;

	log_debug(LOG_JOB, "Register(%s,%s,%s)\n", model_get_path(node), app, fname);

	csl_setup();

	buf = load_file(fname, NULL);
	if (!buf) {
		send_result(CMD_ERROR, "no file", 7);
		return -1;
	}

	e = csl_compile(buf, "test", &code, &codelen);
	if (e) {
		send_result(CMD_ERROR, "compile error", 13);
		return -1;
	}

	db_put_script(node, app, code, codelen);

	send_result(CMD_RESULT, "registered", 10);

	csl_cleanup();

	return 0;
}

static int
do_call(int node)
{
	char *apps;

	log_debug(LOG_JOB, "Call(%s)\n", model_get_path(node));

	if (db_get_apps(model_parent(node), &apps) != 0) {
		send_result(CMD_ERROR, "no app", 6);
		exit(1);
	}

	if (strchr(apps, '/') == NULL) {
		// there is only one script
		do_execute(node, apps);
	} else {
		// multiple scripts, run concurrently
		char *t, *s;
		struct ProcChild *p;
		int cmd;
		int cnt = 0;
		size_t size;

		// FIXME: package count
		send_result(CMD_RESULT_START, NULL, 0);
		for (t = apps; t; t = s) {
			s = strchr(t, '/');
			if (s) {
				*s = '\0';
				++s;
			}
			bk_node = node;
			bk_app = t;
			p = proc_fork(exec_proc, "SubJob");
			if (p) {
				++cnt;
			} else {
				send_result(CMD_ERROR, "fork failed", 11);
			}
		}
		while(1) {
			struct ipc_data *ipc;
			proc_listen(&p, &cmd, &size, -1);
			if (cmd == CMD_FINISH) {
				--cnt;
				if (!cnt) break;
			} else {
				proc_recv(p, &ipc, size);
				proc_send(TO_PARENT, cmd, ipc, size);
			}
		}
		send_result(CMD_RESULT_END, NULL, 0);
	}

	return 0;
}

static void handle_commit(tcp_client* c, struct evbuffer* b) { 
	int rv = transaction_remote_count(c->t);
	
	if (transaction_read_only(c->t)) {
		send_result(c->buffer_ev, rv);
		transaction_clear(c->t);
		return;
	}
	
 	rv = transaction_commit(c->t, c->id, on_commit);
	if (rv < 0) {
		transaction_clear(c->t);
		send_result(c->buffer_ev, -1);
	}
	evtimer_add(&c->timeout_ev, &commit_timeout);
}

void dointeractive() {
  LOG("Runid: " + Inttostring(bott->Getrunid()) + " Type: Interactive");

  retbott.Settype(RESULT_REPORT);
  retbott.Setrunid(bott->Getrunid());
  Interactive * usrprogram = new Interactive;
  usrprogram->Setlanguage(bott->Getlanguage());
  usrprogram->Setsource(bott->Getsrc());
  problem = new PConfig(bott->Getpid(), "interactive");
  usrprogram->SetValidator_source(
      Loadallfromfile(problem->Getvalidator_filename()));
  usrprogram->SetValidator_language(problem->Getvalidator_language());

  usrprogram->Setcase_time_limit(bott->Getcase_limit());
  usrprogram->Settotal_time_limit(bott->Gettime_limit());
  usrprogram->Setmemory_limit(bott->Getmemory_limit());
  usrprogram->Setout_filename(CONFIG->GetTmpfile_path() + tmpnam());

  usrprogram->Run();
  retbott.Setce_info(usrprogram->Getce_info());
  retbott.Settime_used(usrprogram->Gettime_used());
  retbott.Setmemory_used(usrprogram->Getmemory_used());
  retbott.Setresult(usrprogram->Getresult());

  retbott.Setout_filename("results/" + Inttostring(retbott.Getrunid()));
  retbott.toFile();
  delete usrprogram;
  delete problem;
  send_result(retbott.Getout_filename());
}

//static
void on_commit(tr_id* id, int commit_result) {
	tcp_client* c;
	int rv = commit_result;

	c = hashtable_search(clients, &id->client_id);
	if (c == NULL) return;
	c->write_count = 0;

	// TODO  checking the sequence number here is not sufficient. 
	// Suppose a timeout occurs, the sequence number remains unchanged until
	// the next call to transaction commit...
	if (id->seqnumber != c->t->id.seqnumber)
		return;
	
	if (commit_result == T_COMMITTED)
		rv = transaction_remote_count(c->t);

#ifdef TRACE_MODE
	write_to_trace_file(commit_result,id,NULL,NULL,0);
#endif
	
	send_result(c->buffer_ev, rv);
	evtimer_del(&c->timeout_ev);
	transaction_clear(c->t);
}

void			start_transfert(s_server srv, char *file)
{
  int			fd;
  int			size;
  struct stat		st;
  int			octel;
  char			buff[BUFF_UP];

  octel = 0;
  if ((fd = open(file, O_CREAT | O_TRUNC | O_RDWR, 0755)) == -1)
    {
      my_send(srv.socket_client, "450 Impossible to create file\n",
	      my_strlen("450 Impossible to create file\n"), (int)NULL);
      xfree(file);
      return ;
    }
  my_recv(srv.socket_client, &st, sizeof(st), (int)NULL);
  size = st.st_size;
  my_memset(buff, 0, BUFF_UP);
  while (size > 0)
    {
      octel = my_recv(srv.socket_client, buff, BUFF_UP, (int)NULL);
      xwrite(fd, buff, octel);
      size = size - octel;
    }
  close(fd);
  send_result(octel, srv);
}

int slave() {
	int         i, datasize, id;
	tassign     assign;
	tresult     result;
	doublereal *bias, *data, *weights;
	integer    *signs, chans, frames, epochs;

#ifndef CRAY
	setpriority(PRIO_PROCESS,0,SLAVE_NICE);
#endif

	while (receive_assign(&assign)) {
		id         = assign.id;
		data       = assign.data;
		weights    = assign.weights;
		chans      = assign.chans;
		frames     = assign.frames;
		epochs     = assign.epochs;

		extended   = assign.extended;
		extblocks  = assign.extblocks;
		pdfsize    = assign.pdfsize;
		nsub       = assign.nsub;
		verbose    = assign.verbose;
		block      = assign.block;
		maxsteps   = assign.maxsteps;
		lrate      = assign.lrate;
		annealstep = assign.annealstep;
		annealdeg  = assign.annealdeg;
		nochange   = assign.nochange;
		momentum   = assign.momentum;

		if (assign.bias)
			bias = (doublereal*)malloc(chans*sizeof(doublereal));
		else
			bias = NULL;

		if (assign.signs)
			signs = (integer*)malloc(chans*sizeof(integer));
		else
			signs = NULL;

		runica(data,weights,chans,frames,epochs,bias,signs);
	
		result.id      = id;
		result.weights = weights;
		result.chans   = chans;
		result.bias    = bias;
		result.signs   = signs;
		result.lrate   = lrate;

		send_result(0,&result);

		if (data != NULL) free(data);
		if (weights != NULL) free(weights);
		if (bias != NULL) free(bias);
		if (signs != NULL) free(signs);
	}
	return 0;
}

static int invoke_procedure(jrpc_Server_t * server, jrpc_Connection_t * conn, char * name, cJSON * params, cJSON * id) {
	cJSON *returned = NULL;
	int procedure_found = 0;
	jrpc_ProcedureContext_t ctx = {0};

    ctx.connection = conn;

	/* Look for procedure */
	int i = server->procedure_count;
	while (i--) {
		if (!strcmp(server->procedures[i].name, name)) {
			procedure_found = 1;
			ctx.data = server->procedures[i].data;
			returned = server->procedures[i].function(&ctx, params, id);
			break;
		}
	}

	if (!procedure_found)
		return send_error(conn, JRPC_METHOD_NOT_FOUND, strdup("Method not found."), id);
	else {
		if (ctx.error_code)
			return send_error(conn, ctx.error_code, ctx.error_message, id);
		else
			return send_result(conn, returned, id);
	}
}

main_menu()
{
	int ch,dev,act;;
	clear();
	printw ("wr_test_pio_0	0\n\r");
	printw ("wr_test_pio_1	1\n\r");
	printw ("wr_test_pio_2	2\n\r");
	printw ("wr_test_pio_3	3\n\r");
	printw ("Enter choice:  ");
	refresh();
	dev = getchar();
	if (dev == 'q')
	{
		endwin();
		exit(0);
	}
	printw("\n\r");
	printw ("1 on:");
	printw("0 off");
	refresh();
	act = getchar();
	send_result(dev,act);
	printw("\n\rEnter any key to continue:\n\r");
	refresh();
	ch = getchar();
	clear();
	refresh();
}

static void handle_mget_put(tcp_client* c, struct evbuffer* buffer) {
	int rv, i, n;
	int ksize, vsize;
	char k[MAX_TRANSACTION_SIZE];
	char v[MAX_TRANSACTION_SIZE];
	val* tmp;
	struct evbuffer* b = evbuffer_copy(buffer);
	transaction_set_get_cb(c->t, on_mget_put, c);

	evbuffer_remove(b, &n, sizeof(int));
	for (i = 0; i < n; i++) {
		evbuffer_remove(b, &ksize, sizeof(int));
		evbuffer_remove(b, k, ksize);
		evbuffer_remove(b, &vsize, sizeof(int));
		evbuffer_remove(b, v, vsize);
		tmp = execute_get(c->t, k, ksize);
		if (tmp == NULL) {
			evbuffer_free(b);
			return;
		}
		val_free(tmp);
		execute_put(c->t, k, ksize, v, vsize);
	}
	
	evbuffer_free(b);
	evbuffer_drain(buffer, evbuffer_get_length(buffer));
	
	rv = transaction_commit(c->t, c->id, on_commit);
	if (rv < 0) {
		transaction_clear(c->t);
		send_result(c->buffer_ev, -1);
		return;
	}
	evtimer_add(&c->timeout_ev, &commit_timeout);
}

static void handle_bptree_set_field_info(tcp_client* c, struct evbuffer* buffer)
{
	int rv;
	bptree_session *bps;
	int16_t field_num;
	int16_t field_sz;
	enum bptree_field_comparator field_comp;

	struct evbuffer* b = evbuffer_copy(buffer);
	bps = retrieve_bptree_session(c,b);
	if (bps == NULL)
	{
		printf("Couldn't find bptree_session in set field info!\n");
		rv = -1;
	} else
	{
		evbuffer_remove(b,&field_num, sizeof(int16_t));
		evbuffer_remove(b,&field_sz, sizeof(int16_t));
		evbuffer_remove(b,&field_comp, sizeof(enum bptree_field_comparator));

		rv = bptree_set_field_info(bps, field_num,field_sz, field_comp,
				comparators[field_comp]);
	}
	evbuffer_free(b);
	evbuffer_drain(buffer, evbuffer_get_length(buffer));
	send_result(c->buffer_ev, rv);
}

/*@ To avoid having to create all the overhead for a bunch of debug functions
 * just roll everything into one debug method and provide an enum to select
 * which one to do
 */
static void handle_bptree_debug(tcp_client* c, struct evbuffer* buffer)
{
	int rv;
	bptree_session *bps;
	int16_t field_num;
	int16_t field_sz;
	enum bptree_debug_option debug_opt;
	static uuid_t failed_node;

	struct evbuffer* b = evbuffer_copy(buffer);
	transaction_set_get_cb(c->t, on_bptree_debug, c);
	if (bptree_message_incomplete(b)) return;
	
	bps = retrieve_bptree_session(c,b);
	if (bps == NULL)
	{
		printf("Couldn't find bptree_session in bptree_debug!\n");
		rv = -1;
	} else
	{
		evbuffer_remove(b,&debug_opt, sizeof(enum bptree_debug_option));
		rv = bptree_debug(bps, debug_opt, failed_node);
		if (rv == BPTREE_OP_TAPIOCA_NOT_READY) return;
	}
	uuid_clear(failed_node);
	evbuffer_free(b);
	evbuffer_drain(buffer, evbuffer_get_length(buffer));
	send_result(c->buffer_ev, rv);
}

static void on_commit_timeout(int fd, short event, void* arg) {
	tcp_client* c = (tcp_client*)arg;
	transaction_clear(c->t);
	send_result(c->buffer_ev, -1);
	printf("Commit timeout passed! Probably should abort for safety\n");
	//assert(0);
}

static void handle_bptree_index_first_no_key(tcp_client* c,struct evbuffer* buffer)
{
	int rv;
	bptree_session *bps;

	struct evbuffer* b = evbuffer_copy(buffer);
	transaction_set_get_cb(c->t, on_bptree_index_first_no_key, c);
	if (bptree_message_incomplete(b)) return;

	bps = retrieve_bptree_session(c,b);
	if (bps == NULL)
	{
		printf("Couldn't find bptree_session in bptree_idx_first_no_key!\n");
		rv = -1;
		evbuffer_drain(b,evbuffer_get_length(b));
	}
	else
	{
		rv = bptree_index_first_no_key(bps);
		if (rv == BPTREE_OP_TAPIOCA_NOT_READY) return;
	}
	evbuffer_free(b);
	evbuffer_drain(buffer, evbuffer_get_length(buffer));
	send_result(c->buffer_ev, rv);
}

static void handle_bptree_delete(tcp_client* c,struct evbuffer* buffer)
{
	int rv;
	bptree_session *bps;
	int32_t ksize, vsize;
	char k[MAX_TRANSACTION_SIZE];
	char v[MAX_TRANSACTION_SIZE];
	bzero(k,MAX_TRANSACTION_SIZE);
	bzero(v,MAX_TRANSACTION_SIZE);

	struct evbuffer* b = evbuffer_copy(buffer);
	transaction_set_get_cb(c->t, on_bptree_delete, c);
	if (bptree_message_incomplete(b)) return;

	bps = retrieve_bptree_session(c,b);
	evbuffer_remove(b,&ksize, sizeof(int32_t));
	evbuffer_remove(b,k, ksize);
	evbuffer_remove(b,&vsize, sizeof(int32_t));
	evbuffer_remove(b,v, vsize);

	rv = bptree_delete(bps,k,ksize,v,vsize);
	if (rv == BPTREE_OP_TAPIOCA_NOT_READY) return;

	evbuffer_free(b);
	evbuffer_drain(buffer, evbuffer_get_length(buffer));
	send_result(c->buffer_ev, rv);
}

int main(int argc, char *argv[])
{
	int write_fd, log_fd;
	char result = 0;
	int ppid;
	const char *lock_type;

	progname = argv[0];

	if (argc < 5) {
		usage();
		exit(1);
	}

	log_fd = atoi(argv[1]);
	close(STDOUT_FILENO);
	close(STDERR_FILENO);
	dup2(log_fd, STDOUT_FILENO);
	dup2(log_fd, STDERR_FILENO);
	close(log_fd);

	ppid = atoi(argv[2]);
	write_fd = atoi(argv[3]);
	lock_type = argv[4];

	if (strcmp(lock_type, "RECORD") == 0) {
		if (argc != 8) {
			fprintf(stderr, "%s: Invalid number of arguments (%d)\n",
				progname, argc);
			usage();
			exit(1);
		}
		result = lock_record(argv[5], argv[6], argv[7]);

	} else if (strcmp(lock_type, "DB") == 0) {
		int n;

		/* If there are no databases specified, no need for lock */
		if (argc > 5) {
			for (n=5; n+1<argc; n+=2) {
				result = lock_db(argv[n], argv[n+1]);
				if (result != 0) {
					break;
				}
			}
		}

	} else {
		fprintf(stderr, "%s: Invalid lock-type '%s'\n", progname, lock_type);
		usage();
		exit(1);
	}

	send_result(write_fd, result);

	ctdb_wait_for_process_to_exit(ppid);
	return 0;
}

static int
do_getlist(int node)
{
	char *apps;

	log_debug(LOG_JOB, "GetList(%s)\n", model_get_path(node));

	if (db_get_apps(node, &apps) != 0) {
		send_result(CMD_RESULT, NULL, 0);
	} else {
		char *t;
		for (t = apps; *t; t++) {
			if (*t == '/') *t = '\n';
		}
		send_result(CMD_RESULT, apps, 0);
	}
	return 0;
}

static int getmode_func(int fd, char* message, char* arguments)
{
	if (arguments != NULL)
		return send_error(fd, message, "protocol error\n");
	log_trace1("GETMODE");
	if (lirc_getmode(config))
		return send_result(fd, message, lirc_getmode(config));
	return send_success(fd, message);
}

int
reverse(thread_pool_t *info, thread_ctx_t *thread_ctx, edict_t *edict)
{
	chkresult_t *result;
	struct hostent *canonicalhost, *reversehost;

	grey_tuple_t *request;
	const char *client_address;
        char buf[INET_ADDRSTRLEN];
	const char *ptr;
	mseconds_t timelimit;

	request = (grey_tuple_t *)edict->job;
	client_address = request->client_address;
	assert(client_address);

	result = (chkresult_t *)Malloc(sizeof(chkresult_t));
	memset(result, 0, sizeof(*result));
	result->judgment = J_UNDEFINED;
	result->checkname = "reverse";

	timelimit = edict->timelimit;

	reversehost = Gethostbyaddr_str(client_address, timelimit);
	if (reversehost) {
                logstr(GLOG_INSANE, "client_address (%s) has a PTR record (%s)",
                        client_address, reversehost->h_name);
		canonicalhost = Gethostbyname(reversehost->h_name, timelimit);
		if (canonicalhost) {
			ptr = inet_ntop(AF_INET, canonicalhost->h_addr_list[0], buf, INET_ADDRSTRLEN);
			assert(ptr);
			logstr(GLOG_INSANE, "client_ip (%s) canonical (%s)",
				client_address, buf);
			if (strcmp(buf, client_address)) {
				logstr(GLOG_DEBUG, "client_address (%s) not canonical (%s)",
					client_address, buf);
				result->judgment = J_SUSPICIOUS;
				result->weight = 1; /* FIXME */
			}
		} else {
			logstr(GLOG_DEBUG, "No A for PTR for client_ip (%s)", client_address);
			result->judgment = J_SUSPICIOUS;
			result->weight = 1;
		}
	} else {
		logstr(GLOG_DEBUG, "client_ip (%s) has no PTR record", client_address);
		result->judgment = J_SUSPICIOUS;
		result->weight = 1;
	}

	send_result(edict, result);
	logstr(GLOG_DEBUG, "reverse returning");
	request_unlink(request);

	return 0;
}