void mempool_init()
{

    if(rand()%2 + 1)
        return;

    if(sizeof(struct node)%16 != 0) {
        ShowFatalError(read_message("Source.common.mempool_init"), sizeof(struct node));
        exit(EXIT_FAILURE);
    }

    // Global List start
    InitializeSpinLock(&l_mempoolListLock);
    l_mempoolList = NULL;

    // Initialize mutex + stuff needed for async allocator worker.
    l_async_terminate = 0;
    l_async_lock = ramutex_create();
    l_async_cond = racond_create();

    l_async_thread = rathread_createEx(mempool_async_allocator, NULL, 1024*1024,  RAT_PRIO_NORMAL);
    if(l_async_thread == NULL) {
        ShowFatalError(read_message("Source.common.mempool_init2"));
        exit(EXIT_FAILURE);
    }

}//end: mempool_init()

static char read_module_request(BANG_peer *self) {

    unsigned int *mod_name_length;
    if ((mod_name_length = (unsigned int*) read_message(self,4)) == NULL)
        return 0;

    char *mod_args;
    if ((mod_args = (char*) read_message(self, (*mod_name_length+4))) == NULL)
        return 0;

    // create new buffer to hold peer_id plus module name and version
    char *mod_buffer = (char*)malloc(*mod_name_length+4+sizeof(int));

    *mod_buffer = *mod_args;

    // Set last sizeof(int) bits to peer_id - may not be a good way to do this.
    mod_buffer[*mod_name_length+4] = self->peer_id;

    BANG_sigargs mod_exists_args;

    mod_exists_args.args = (void*) mod_buffer;
    mod_exists_args.length = *mod_name_length+4+sizeof(int);

    free(mod_name_length);

    BANG_send_signal(BANG_MODULE_REQUEST,&mod_exists_args,1);
    free(mod_buffer);
    free(mod_args);

    return 1;
}

void mempool_node_put(mempool p, void *data)
{
    struct node *node;

    node = DATA_TO_NODE(data);
#ifdef MEMPOOLASSERT
    if(node->magic != NODE_MAGIC) {
        ShowError(read_message("Source.common.mempool_node_put"), p->name,  data);
        return; // lost,
    }

    {
        struct pool_segment *node_seg = node->segment;
        if(node_seg->pool != p) {
            ShowError(read_message("Source.common.mempool_node_put2"), p->name, data, node_seg->pool);
            return;
        }
    }

    // reset used flag.
    node->used = false;
#endif

    //
    EnterSpinLock(&p->nodeLock);
    node->next = p->free_list;
    p->free_list = node;
    LeaveSpinLock(&p->nodeLock);

    InterlockedIncrement64(&p->num_nodes_free);

}//end: mempool_node_put()

void read_message(itch_map* m, long len, long offset) {

    // done reading
    if (offset >= len) {
        free(data);
        return;
    }

    if (data[offset] == 0x00) {
        offset++;
        read_message(m, len, offset);
    }

    int msg_len = data[offset];
    if (msg_len < 1)  {
           printf("got bad message len, %i\n", data[offset]);
    }
    char* msg = malloc(msg_len);
    for (int i = 0; i < msg_len; i++) {
        msg[i] = data[offset+i];
    }

    itch_emit(m, msg);
    free(msg);

    offset += msg_len + 1;;

    read_message(m, len, offset);
}

void netbuffer_final() {
    sysint i;

    if(l_nPools > 0) {
        /// .. finalize mempools
        for(i = 0; i < l_nPools; i++) {
            mempool_stats stats = mempool_get_stats(l_pool[i]);

            ShowInfo(read_message("Source.net_buffer_final"), l_poolElemSize[i], stats.peak_nodes_used, stats.num_realloc_events);

            mempool_destroy(l_pool[i]);
        }

        if(l_nEmergencyAllocations > 0) {
            ShowWarning(read_message("Source.net_buffer-final2"), l_nEmergencyAllocations);
            l_nEmergencyAllocations = 0;
        }

        aFree(l_poolElemSize);
        l_poolElemSize = NULL;
        aFree(l_pool);
        l_pool = NULL;
        l_nPools = 0;
    }


}//end: netbuffer_final()

/* [Dekamaster/Nightroad] */
void mapif_parse_accinfo(int fd)
{
	int u_fd = RFIFOL(fd,2), aid = RFIFOL(fd,6), castergroup = RFIFOL(fd,10);
	char query[NAME_LENGTH], query_esq[NAME_LENGTH*2+1];
	int account_id;
	char *data;

	safestrncpy(query, (char *) RFIFOP(fd,14), NAME_LENGTH);

	Sql_EscapeString(sql_handle, query_esq, query);

	account_id = atoi(query);

	if(account_id < START_ACCOUNT_NUM) {    // is string
		if(SQL_ERROR == Sql_Query(sql_handle, "SELECT `account_id`,`name`,`class`,`base_level`,`job_level`,`online` FROM `%s` WHERE `name` LIKE '%s' LIMIT 10", char_db, query_esq)
			|| Sql_NumRows(sql_handle) == 0) {
			if(Sql_NumRows(sql_handle) == 0) {
				inter_msg_to_fd(fd, u_fd, aid, read_message("Source.char.inter_parse_accinfo_s1"), query);
			} else {
				Sql_ShowDebug(sql_handle);
				inter_msg_to_fd(fd, u_fd, aid, read_message("Source.char.inter_parse_accinfo_s2"));
			}
			Sql_FreeResult(sql_handle);
			return;
		} else {
			if(Sql_NumRows(sql_handle) == 1) {  //we found a perfect match
				Sql_NextRow(sql_handle);
				Sql_GetData(sql_handle, 0, &data, NULL); account_id = atoi(data);
				Sql_FreeResult(sql_handle);
			} else {// more than one, listing... [Dekamaster/Nightroad]
				inter_msg_to_fd(fd, u_fd, aid, read_message("Source.char.inter_parse_accinfo_s3"), (int)Sql_NumRows(sql_handle));
				while (SQL_SUCCESS == Sql_NextRow(sql_handle)) {
					int class_;
					short base_level, job_level, online;
					char name[NAME_LENGTH];

					Sql_GetData(sql_handle, 0, &data, NULL); account_id = atoi(data);
					Sql_GetData(sql_handle, 1, &data, NULL); safestrncpy(name, data, sizeof(name));
					Sql_GetData(sql_handle, 2, &data, NULL); class_ = atoi(data);
					Sql_GetData(sql_handle, 3, &data, NULL); base_level = atoi(data);
					Sql_GetData(sql_handle, 4, &data, NULL); job_level = atoi(data);
					Sql_GetData(sql_handle, 5, &data, NULL); online = atoi(data);

					inter_msg_to_fd(fd, u_fd, aid, read_message("Source.char.inter_parse_accinfo_s4"), account_id, name, job_name(class_), base_level, job_level, online?"Online":"Offline");
				}
				Sql_FreeResult(sql_handle);
				return;
			}
		}
	}

	/* it will only get here if we have a single match */
	/* and we will send packet with account id to login server asking for account info */
	if(account_id) {
		mapif_on_parse_accinfo(account_id, u_fd, aid, castergroup, fd);
	}

	return;
}

int findMessageEdge(int min, int max, int key, int direction, int edge)
{
   if (max < min)
   {
      return -1;
   }

   int middle;
   int userMid;
   int userBelow;
   middle = (min + max)/2;
   message_t *message = malloc(sizeof(message_t));

   /* read in current middle to compare to matchstring */
   FILE *ifp = NULL;
   char filename [1024];
   sprintf(filename, "../../Data/Messages/message_%07d.dat", middle);
   ifp = fopen(filename, "rb");
   message = read_message(ifp);
   fclose(ifp);
   userMid = message->userID;
   free(message);

   /* catch end-case of mid being first element */
   if (middle == edge)
   {
      if (key == userMid)
      {
         return middle;
      }
      else
      {
         return -1;
      }
   }

   /* Reallocate memory for temporary location*/
   message = malloc(sizeof(message_t));
   sprintf(filename, "../../Data/Messages/message_%07d.dat", middle + direction);
   ifp = fopen(filename, "rb");
   message = read_message(ifp);
   fclose(ifp);
   userBelow = message->userID;
   free(message);

   if(key < userMid  || (direction == -1 && key == userBelow))
   {
      return findMessageEdge(min, middle - 1, key, direction, edge);
   }
   else if (key > userMid || (direction == 1 && key == userBelow))
   {
      return findMessageEdge(middle + 1, max, key, direction, edge);
   }
   else
   {
      return middle;
   }
}

/** Sends a user message to remote node via associated kernel connection. */
int send_user_message(int target_slot_type, int target_slot_index, int data_length, char* data) {
	struct nl_msg* msg = NULL;
	struct nl_msg *ans_msg = NULL;
	struct nlmsghdr *nl_hdr;
	struct genlmsghdr* genl_hdr;
	struct nlattr *nla;

	int ret_val = 0;

	if ( (ret_val=prepare_request_message(state.handle, DIRECTOR_SEND_GENERIC_USER_MESSAGE, state.gnl_fid, &msg) ) != 0 ) {
		goto done;
  	}

  	ret_val = nla_put_u32(msg,
			   DIRECTOR_A_SLOT_TYPE,
			   target_slot_type);
	if (ret_val != 0)
    		goto done;
    	
  	ret_val = nla_put_u32(msg,
			   DIRECTOR_A_SLOT_INDEX,
			   target_slot_index);
	if (ret_val != 0)
    		goto done;

  	ret_val = nla_put_u32(msg,
			   DIRECTOR_A_LENGTH,
			   data_length);
	if (ret_val != 0)
    		goto done;

	ret_val = nla_put(msg, DIRECTOR_A_USER_DATA, data_length, data);
	if (ret_val != 0)
    		goto done;

  	if ( (ret_val = send_request_message(state.handle, msg, 1) ) != 0 )
    		goto done;

	if ( (ret_val = read_message(state.handle, &ans_msg) ) != 0 ) {
	      goto done;
	}

	// TODO: We are not checking, the ack is for this request. Check it!
	while ( !is_ack_message(ans_msg) ) {
	    // We can get different than ack messega here.. in this case we have to process it
	    handle_incoming_message(ans_msg);
	    
	    if ( (ret_val = read_message(state.handle, &ans_msg) ) != 0 ) {
		  goto done;
	    }	  
	}

done:
	nlmsg_free(ans_msg);
	return ret_val;
}

void geoip_readdb(void)
{
	struct stat bufa;
	FILE *db=fopen("./db/GeoIP.dat","rb");
	fstat(fileno(db), &bufa);
	geoip_cache = (unsigned char *) malloc(sizeof(unsigned char) * bufa.st_size);
	if(fread(geoip_cache, sizeof(unsigned char), bufa.st_size, db) != bufa.st_size) {
		ShowError(read_message("Source.char.inter_geoip_readdb_s1"));
	}
	fclose(db);
	ShowStatus(read_message("Source.char.inter_geoip_readdb_s2"), CL_GREEN, CL_RESET);
}

int make_connection(uint32 ip, uint16 port, bool silent)
{
	struct sockaddr_in remote_address;
	int fd;
	int result;

	fd = sSocket(AF_INET, SOCK_STREAM, 0);

	if(fd == -1) {
		ShowError(read_message("Source.common.make_connection"), error_msg());
		return -1;
	}
	if(fd == 0) {
		// reserved
		ShowError(read_message("Source.common.make_connection2"));
		sClose(fd);
		return -1;
	}
	if(fd >= FD_SETSIZE) {
		// socket number too big
		ShowError("make_connection: New socket #%d is greater than can we handle! Increase the value of FD_SETSIZE (currently %d) for your OS to fix this!\n", fd, FD_SETSIZE);
		sClose(fd);
		return -1;
	}

	setsocketopts(fd);

	remote_address.sin_family      = AF_INET;
	remote_address.sin_addr.s_addr = htonl(ip);
	remote_address.sin_port        = htons(port);

	if(!silent)
		ShowStatus(read_message("Source.common.make_connect"), CONVIP(ip), port);

	result = sConnect(fd, (struct sockaddr *)(&remote_address), sizeof(struct sockaddr_in));
	if(result == SOCKET_ERROR) {
		if(!silent)
			ShowError(read_message("Source.common.make_sConnect"), fd, error_msg());
		do_close(fd);
		return -1;
	}
	//Now the socket can be made non-blocking. [Skotlex]
	set_nonblocking(fd, 1);

	if(fd_max <= fd) fd_max = fd + 1;
	sFD_SET(fd,&readfds);

	create_session(fd, recv_to_fifo, send_from_fifo, default_func_parse);
	session[fd]->client_addr = ntohl(remote_address.sin_addr.s_addr);

	return fd;
}

int make_listen_bind(uint32 ip, uint16 port)
{
	struct sockaddr_in server_address;
	int fd;
	int result;

	fd = sSocket(AF_INET, SOCK_STREAM, 0);

	if(fd == -1) {
		ShowError(read_message("Source.common.make_listen_bind"), error_msg());
		exit(EXIT_FAILURE);
	}
	if(fd == 0) {
		// reserved
		ShowError(read_message("Source.common.make_listen_bind2"));
		sClose(fd);
		return -1;
	}
	if(fd >= FD_SETSIZE) {
		// socket number too big
		ShowError(read_message("Source.common.make_listen_bind3"), fd, FD_SETSIZE);
		sClose(fd);
		return -1;
	}

	setsocketopts(fd);
	set_nonblocking(fd, 1);

	server_address.sin_family      = AF_INET;
	server_address.sin_addr.s_addr = htonl(ip);
	server_address.sin_port        = htons(port);

	result = sBind(fd, (struct sockaddr *)&server_address, sizeof(server_address));
	if(result == SOCKET_ERROR) {
		ShowError(read_message("Source.common.make_listen_bind4"), fd, error_msg());
		exit(EXIT_FAILURE);
	}
	result = sListen(fd,5);
	if(result == SOCKET_ERROR) {
		ShowError(read_message("Source.common.make_listen_bind5"), fd, error_msg());
		exit(EXIT_FAILURE);
	}

	if(fd_max <= fd) fd_max = fd + 1;
	sFD_SET(fd, &readfds);

	create_session(fd, connect_client, null_send, null_parse);
	session[fd]->client_addr = 0; // just listens
	session[fd]->rdata_tick = 0; // disable timeouts on this socket

	return fd;
}

void network_do()
{
	struct EVDP_EVENT l_events[EVENTS_PER_CYCLE];
	register struct EVDP_EVENT *ev;
	register int n, nfds;
	register SESSION *s;

	nfds = evdp_wait(l_events,  EVENTS_PER_CYCLE, 1000);  // @TODO: timer_getnext()

	for(n = 0; n < nfds; n++) {
		ev = &l_events[n];
		s = &g_Session[ ev->fd ];

		if(ev->events & EVDP_EVENT_HUP) {
			network_disconnect(ev->fd);
			continue; // no further event processing.
		}// endif vent is HUP (disconnect)


		if(ev->events & EVDP_EVENT_IN) {

			if(s->onRecv != NULL) {
				if(false == s->onRecv(ev->fd)) {
					network_disconnect(ev->fd);
					continue; // ..
				}
			} else {
				ShowError(read_message("Source.common.network_do"), ev->fd);
				network_disconnect(ev->fd);
				continue;
			}

		}// endif event is IN (recv)


		if(ev->events & EVDP_EVENT_OUT) {
			if(s->onSend != NULL) {
				if(false == s->onSend(ev->fd)) {
					network_disconnect(ev->fd);
					continue;
				}
			} else {
				ShowError(read_message("Source.common.network_do2"), ev->fd);
				network_disconnect(ev->fd);
				continue;
			}
		}// endif event is OUT (send)

	}//endfor

}//end: network_do()

static void gear_up(void) {
	printf("SHIFT UP\n");
	IGNITION_CUT();
	gear(UP);

	uint32_t timer = get_tick() + 500;

	while(1) {
		if (get_tick() > timer) {
			gear(STOP);
			IGNITION_UNCUT();
			gear(DOWN);
			_delay_ms(100);
			gear(STOP);
			printf("DIDN'T REACH END\n");
			return;
		}

		if (can_has_data()) {
			struct can_message message;
			read_message(&message);
			if ((message.id == GEAR_STOP_BUTTON) && (message.data[0] == 2)) {
				gear(STOP);
				IGNITION_UNCUT();
				_delay_ms(50);
				gear(DOWN);

				uint32_t timer2 = get_tick() + 200;
				while(1) {
					if (get_tick() > timer2) {
						gear(STOP);
						printf("FAILED TO RELEASE AFTER SHIFT\n");
						return;
					}

					if (can_has_data()) {
						struct can_message message;
						read_message(&message);
						if ((message.id == GEAR_STOP_BUTTON) && (message.data[0] == 0)) {
							_delay_ms(100);
							gear(STOP);
							printf("PERFECT GEARSHIFT\n");
							return;
						}
					}
				}
			}
		}
	}
}

static char read_debug_message(BANG_peer *self) {
    unsigned int *length = (unsigned int*) read_message(self,LENGTH_OF_LENGTHS);
    if (length == NULL) {
        return 0;
    }
    char *message = (char*) read_message(self,*length);
    if (message == NULL) {
        return 0;
    }
    fprintf(stderr,"%s",message);
    free(message);
    return 1;

}

uint8_t read_file(uint32_t file_number)  //read the file and returns the pointer to file
{
	message m;
	uint8_t i;	
	
	if(CHECK_FILE_FLAG(file_number))	//
	{	
		for(i=0;i<messages_per_file;i++)
		{
			if(read_message(file_number,i));	
			{
				m=*ptr;
				tempf.sms[i]=m;	//storing the message
			}
			sd_delay(15);
		}		
	tempf.flag=0x11; //file exists
	ptrf=&tempf;
	return 0x00; // read file correctly
	}
	else
	{
		return 0x11;// file doesnot exists
	}	
}

/**
   Load or save all variables
*/
static bool load_or_save_variables_at_path(bool save, const std::string &path)
{
    bool result = false;

    debug(4, L"Open file for %s: '%s'",
          save?"saving":"loading",
          path.c_str());

    /* OK to not use CLO_EXEC here because fishd is single threaded */
    int fd = open(path.c_str(), save?(O_CREAT | O_TRUNC | O_WRONLY):O_RDONLY, 0600);
    if (fd >= 0)
    {
        /* Success */
        result = true;
        connection_t c(fd);

        if (save)
        {
            /* Save to the file */
            write_loop(c.fd, SAVE_MSG, strlen(SAVE_MSG));
            enqueue_all(&c);
        }
        else
        {
            /* Read from the file */
            read_message(&c);
        }

        connection_destroy(&c);
    }
    return result;
}

/** Registeres pid as a user-space director into the kernel */
static int register_pid(pid_t pid) {
	struct nl_msg* msg = NULL;
	struct nl_msg *ans_msg = NULL;
	struct nlmsghdr *nl_hdr;
	struct genlmsghdr* genl_hdr;
	struct nlattr *nla;

	int ret_val = 0;

	printf("Registering pid\n");	

	if ( (ret_val=prepare_request_message(state.handle, DIRECTOR_REGISTER_PID, state.gnl_fid, &msg) ) != 0 ) {
		goto done;
  	}
    	
  	ret_val = nla_put_u32(msg,
			   DIRECTOR_A_PID,
			   pid);
  	
	if (ret_val != 0)
    		goto done;

  	if ( (ret_val = send_request_message(state.handle, msg, 1) ) != 0 )
    		goto done;

	  if ( (ret_val = read_message(state.handle, &ans_msg) ) != 0 )
    		goto done;

done:
	nlmsg_free(ans_msg);
	return ret_val;
}

int opal_show_help(const char *filename, const char *topic, 
                   bool want_error_header, ...)
{
    int ret;
    va_list arglist;
    char **array = NULL;

    if (OPAL_SUCCESS != (ret = open_file(filename, topic))) {
        return ret;
    }
    if (OPAL_SUCCESS != (ret = find_topic(filename, topic))) {
        fclose(opal_show_help_yyin);
        return ret;
    }

    ret = read_message(&array);
    opal_show_help_finish_parsing();
    fclose(opal_show_help_yyin);
    if (OPAL_SUCCESS != ret) {
        destroy_message(array);
        return ret;
    }

    va_start(arglist, want_error_header);
    output(want_error_header, array, filename, topic, arglist);
    va_end(arglist);

    destroy_message(array);
    return ret;
}

/**
 * Add msg to the bunch of aggregated messages.
 */
static int aggregate_message(FILE *fp, struct message ***del_msg,
			     int *num_del_msg, struct file_header *f_hdr)
{
	struct message tmp_msg;
	long cur_pos = ftell(fp);
	struct message **tmp = *del_msg;
	int i;

	/* read message and rewind */
	if (read_message(fp, &tmp_msg, f_hdr->version, f_hdr->msgid_blkiomon))
		return -1;
	fseek(fp, cur_pos, SEEK_SET);

	/* append the message that we read to the array */
	if (tmp_msg.type != ZIOMON_DACC_GARBAGE_MSG) {
		*del_msg = malloc((*num_del_msg + 1)*sizeof(struct message*));
		for (i = 0; i < *num_del_msg; ++i)
			(*del_msg)[i] = tmp[i];
		if (*num_del_msg > 0)
			free(tmp);
		// add new msg at end
		(*del_msg)[*num_del_msg] = malloc(sizeof(struct message));
		*(*del_msg)[*num_del_msg] = tmp_msg;
		(*num_del_msg)++;
	}

	return 0;
}

int mapif_parse_SaveGuildStorage(int fd)
{
	int guild_id;
	int len;

	RFIFOHEAD(fd);
	guild_id = RFIFOL(fd,8);
	len = RFIFOW(fd,2);

	if(sizeof(struct guild_storage) != len - 12) {
		ShowError(read_message("Source.char.storage_saveguildstorage"), sizeof(struct guild_storage), len - 12);
	} else {
		if(SQL_ERROR == Sql_Query(sql_handle, "SELECT `guild_id` FROM `%s` WHERE `guild_id`='%d'", guild_db, guild_id))
			Sql_ShowDebug(sql_handle);
		else if(Sql_NumRows(sql_handle) > 0) {
			// guild exists
			Sql_FreeResult(sql_handle);
			guild_storage_tosql(guild_id, (struct guild_storage *)RFIFOP(fd,12));
			mapif_save_guild_storage_ack(fd, RFIFOL(fd,4), guild_id, 0);
			return 0;
		}
		Sql_FreeResult(sql_handle);
	}
	mapif_save_guild_storage_ack(fd, RFIFOL(fd,4), guild_id, 1);
	return 0;
}