Сокеты, одновременная работа с несколькими клиентами !!!

#include <iostream>
#include <string>
#include <map>
//#include "hiredis.h"
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/shared_array.hpp>
#include <boost/thread.hpp>
 
#include <boost/tokenizer.hpp>
 
/***************************************************************************/
 
#define count_checkbox 1000
 
// максимальная длина принимаемого сообщения.
static const size_t message_len = 10000;
 
typedef boost::shared_ptr<boost::asio::ip::tcp::socket> socket_ptr;
 
// начало приема запросов на подключение.
void start_accept(
   boost::asio::ip::tcp::acceptor* acceptor
);
 
// создаем асинхронную операцию аксепта.
void on_accept(
   boost::asio::ip::tcp::acceptor* acceptor,
   socket_ptr socket
);
 
// вызовется при подключении.
void on_accept_handler(
   boost::asio::ip::tcp::acceptor* acceptor,
   socket_ptr socket,
   const boost::system::error_code& error // объект ошибки
);
 
// вызовется при завершении асинхронной записи.
void on_async_read(
   socket_ptr socket,
   boost::shared_array<char> message, // shared_array мы сюда передаем для
                                      // того, чтоб не дать ему удалить данные которыми он
                                      // владеет до завершения асинхронной записи.
   const boost::system::error_code& error // объект ошибки
);
 
void on_async_write(
  const boost::system::error_code& error,
  std::size_t bytes_transferred
);
 
boost::asio::io_service ios;
 
/***************************************************************************/
 
//redisContext *c;
//redisReply *reply;
 
//Поисковая база
int count = 0;
 
typedef struct {
        unsigned int id;
        unsigned short *interests;
        unsigned int bithday;
        bool male;
} one_element;
 
 
typedef std::map <std::string, one_element> search_elements;
 
std::map <unsigned int, search_elements> search_elements_sort;
 
std::map <unsigned int, unsigned int> sort_elements;
 
 
std::string int2str(int i) {
 
    std::stringstream ss;
    ss<<i;
    std::string str;
    ss>>str;
    return str;
}
 
/*
std::vector<std::string> explode(char *sep, std::string src)
{
    std::vector<std::string> output;
    boost::char_separator<char> separator(sep);
    boost::tokenizer<boost::char_separator<char> > tokens(src, separator);
    boost::tokenizer<boost::char_separator<char> >::iterator token_iter;
 
    for (token_iter = tokens.begin(); token_iter != tokens.end(); token_iter++)
    output.push_back(*token_iter);
 
    return output;
}
 
std::string implode(char sep[1], std::vector<std::string> src)
{
    std::ostringstream output;
    std::vector<std::string>::iterator src_iter;
 
    for (src_iter = src.begin(); src_iter != src.end(); src_iter++) {
        output << *src_iter;
        if (src_iter != src.end()-1) {
            output << sep;
        }
    }
    return output.str();
}
 
 
 
void load_data_from_redis() {
 
 
 
    reply = (redisReply*)redisCommand(c, "FLUSHALL");
    freeReplyObject(reply);
 
    int all_insert = 10000;
    redisCommand(c, "SET users:count %i", all_insert);
    for (int i=1; i<=all_insert; i++) {
        int sort = (rand() % 501) + 100;
        int bithday = 100000;
        int male = rand() % 2;
        std::vector<std::string> v;
        for (int g = 1; g <= count_checkbox; g++) {
            if(rand() % 30 == 0) {
                v.push_back(int2str(g));
            }
        }
        std::string interests = implode((char*)" ! ", v);
        char buf[interests.length()];
        std::strcpy(buf, interests.c_str());
 
        redisCommand(c, "HMSET users:info:%i id %i sort %i status 1 bithday %i male %i interests \"%s\"", i, i, sort, bithday, male, buf);
 
    }
 
 
    reply = (redisReply*)redisCommand(c, "GET users:count");
    int all_count_check = atoi(reply->str);
    freeReplyObject(reply);
    for (int i=1; i <= all_count_check; i++) {
        reply = (redisReply*)redisCommand(c, "HGETALL users:info:%i", i);
        if (reply->elements > 0) {
            int j_s = reply->elements;
            int sort;
            int status;
            one_element tmp;
            for (int j=0; j<j_s; j+=2) {
                if (reply->element[j]->str == (std::string)"id") {
                    tmp.id = atoi(reply->element[j+1]->str);
                } else if (reply->element[j]->str == (std::string)"sort") {
                    sort = atoi(reply->element[j+1]->str);
                } else if (reply->element[j]->str == (std::string)"interests") {
                    std::string interests = reply->element[j+1]->str;
                    std::vector<std::string> v = explode((char*)" ! ",  interests.substr(1, interests.length()-2));
                    std::vector<std::string>::iterator src_iter;
                    tmp.interests = (unsigned short*)malloc(sizeof(unsigned short)*v.size());
                    int c = 0;
                    for (src_iter = v.begin(); src_iter != v.end(); src_iter++) {
                        std::string str = *src_iter;
                        tmp.interests[c] = atoi(str.c_str());
                        c++;
                    }
                } else if (reply->element[j]->str == (std::string)"status") {
                    status = atoi(reply->element[j+1]->str);
                } else if (reply->element[j]->str == (std::string)"bithday") {
                    tmp.bithday = atoi(reply->element[j+1]->str);
                } else if (reply->element[j]->str == (std::string)"male") {
                    if (reply->element[j+1]->str == (std::string)"1") {
                        tmp.male = true;
                    } else {
                        tmp.male = false;
                    }
                }
            }
            if (status == 1) {
                count++;
                search_elements_sort[sort][int2str(tmp.id)] = tmp;
                sort_elements[tmp.id] = sort;
            }
            free(tmp.interests);
        }
        freeReplyObject(reply);
    }
 
}
 
 
std::vector<std::string> make_search() {
 
    std::vector<std::string> result;
    std::map <unsigned int, search_elements> :: iterator cur;
    search_elements :: iterator cur_in;
 
    for (cur = search_elements_sort.begin(); cur != search_elements_sort.end(); cur++) {
        for (cur_in = cur->second.begin(); cur_in != cur->second.end(); cur_in++) {
            if (cur_in->second.male == true) {
                result.push_back(cur_in->first);
            }
        }
    }
 
    return result;
}
 
*/
std::string execute_command(int type, std::string message) {
 
    std::string result;
    /*
    //Здесь обработка я ее закоментировал
    if (type == 1) {
        std::vector<std::string> search = make_search();
        std::vector<std::string> :: iterator cur_r;
        for (cur_r = search.begin(); cur_r != search.end(); cur_r++) {
            result += *cur_r;
            if (cur_r != search.end()-1) {
                result += ",";
            }
        }
    }
    */
    sleep(10);
    return result;
 
}
 
 
int main() {
 
    /* Коннект к редису, и загрузка данных */
/*
    struct timeval timeout = { 1, 500000 }; // 1.5 seconds
    c = redisConnectWithTimeout((char*)"127.0.0.1", 6379, timeout);
    if (c->err) {
        printf("Connection to Redis error: %s\n", c->errstr);
        exit(1);
    }
 
    printf("Start data loading from Redis...\n");
 
    load_data_from_redis();
 
    printf("Data is loaded, now you can send commands...\n");
 
*/
 
    /* Открываем сокеты и формируем пул-потоки */
   static const char* ip = "127.0.0.1";
   static const boost::uint16_t port = 44550;
 
   // объект реализующий "вечную задачу".
   boost::shared_ptr<boost::asio::io_service::work> work(new boost::asio::io_service::work(ios));
 
   // определяем кол-во ядер.
   size_t cpu_count = 2;//boost::thread::hardware_concurrency();
 
   // создаем кол-во рабочих потоков равное кол-ву ядер.
   boost::thread_group work_threads;
   for ( size_t idx = 0; idx < cpu_count; ++idx ) {
      work_threads.create_thread(boost::bind(&boost::asio::io_service::run, &ios));
   }
 
   // объект представляющий адрес для TCP соединений.
   boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(ip), port);
 
   // объект акксептора соединений.
   boost::asio::ip::tcp::acceptor acceptor(ios, endpoint);
 
   // запускаем асинхронный прием запросов на подключение
   start_accept(&acceptor);
 
 
   // сбрасываем "вечную задачу".
   work.reset();
 
   // работаем пока не завершаться все потоки.
   work_threads.join_all();
 
   // завершаемся.
   return 0;
}
 
/***************************************************************************/
 
// создаем асинхронную операцию аксепта
void start_accept(
   boost::asio::ip::tcp::acceptor* acceptor)
{
 
    std::cout << "WAIT NEW CONNECTION... \n";
   // создаем объект сокета который будет связан с принятым подключением.
   socket_ptr socket(new socket_ptr::element_type(acceptor->get_io_service()));
 
   // создаем операци. асинхронного аксепта.
   acceptor->async_accept(
      *socket,
      boost::bind(
         &on_accept_handler,
         acceptor,
         socket,
         boost::asio::placeholders::error
      )
   );
}
 
 
 
// вызовется при приеме нового подключения.
void on_accept_handler(
   boost::asio::ip::tcp::acceptor* acceptor,
   socket_ptr socket,
   const boost::system::error_code& error)
{
   if ( !error ) {
      // создаем shared_array который должен жить вплоть до окончании асинхронного чтения.
      // для этого, message передаем как аргумент в функциональный объект, для того, чтоб инкрементировать счетчик ссылок.
 
      boost::shared_array<char> message_read(new char[message_len]);
      //message_read.reset(new char[message_len]); - пробовал очищать буфер, толку никакого
 
      // для принятого соединения запускаем операцию асинхронного чтения.
 
      boost::asio::async_read(
         *socket,
         boost::asio::buffer(message_read.get(), message_len),
         boost::asio::transfer_at_least(2),
         boost::bind(
            &on_async_read,
            socket,
            message_read,
            boost::asio::placeholders::error
         )
 
       );
 
 
 
   } else {
      std::cout << "on_accept_handler() error: " << error.message() << std::endl;
   }
   // снова запускаем асинхронный прием запросов на подключение
   start_accept(acceptor);
}
 
/***************************************************************************/
 
// вызовется при завершении асинхронного чтения.
 
// примечание: аргумент message и socket, разрушаться при выходе из функции.
void on_async_read(
   socket_ptr socket,
   boost::shared_array<char> message_read,
   const boost::system::error_code& error)
{
   if ( !error ) {
    std::cout << "received message: " << message_read.get() << std::endl;
 
    std::string answer = execute_command(1, message_read.get());
 
       boost::asio::async_write(
         *socket,
         boost::asio::buffer(answer, answer.length()),
         &on_async_write
      );
 
   } else {
      std::cout << "on_async_read() error: " << error.message() << std::endl;
   }
}
 
 
void on_async_write(const boost::system::error_code& error, std::size_t bytes_transferred) {
 
///
 
};
 
/***************************************************************************/