void SlotOnNewDataHandler(net::ClientId client_id, net::ServerId server_id, common::Byteset data)
    {
        for (unsigned int n = 0; n < data.GetSize(); n++)
        {
            this->buffer_.push_back(data[n]);
        }

        while (this->buffer_.size() >= 8)
        {
            std::string command = "";
            command += (char)this->buffer_[0];
            command += (char)this->buffer_[1];
            command += (char)this->buffer_[2];
            command += (char)this->buffer_[3];

            std::string payload_length_str = "";
            payload_length_str += (char)this->buffer_[4];
            payload_length_str += (char)this->buffer_[5];
            payload_length_str += (char)this->buffer_[6];
            payload_length_str += (char)this->buffer_[7];
            
            unsigned int payload_length = boost::lexical_cast<unsigned int>(payload_length_str);
            
            if (this->buffer_.size() - 8 < payload_length)
            {
                return;
            }
         
            std::string payload = "";
            for (unsigned int n = 8; n < payload_length + 8; n++)
            {
                payload += (char)this->buffer_[n];
            }
            
            this->buffer_.erase(this->buffer_.begin(), this->buffer_.begin() + payload_length + 8);
            
            if (command == "TEXT")
            {
                std::cout << payload << std::flush;
            }
            else if (command == "PROM")
            {
                this->prompt_ = payload;
                on_message_condition.notify_all();
            }
            else if (command == "COMP")
            {
                autocomplete_list.clear();
                
                if (payload.length() > 0)
                {
                    boost::algorithm::split(autocomplete_list, payload, boost::is_any_of("\n"), boost::algorithm::token_compress_on);
                }
                
                on_message_condition.notify_all();
            }
            else
            {
                std::cerr << "Could not parse package." << std::endl;
                finish = true;
                on_message_condition.notify_all();
                break;
            }
        }
    }

int main(int argc, char **argv)
{
    // Signal handlers
    signal(SIGTERM, Handler);
    signal(SIGINT, Handler);
    signal(SIGQUIT, Handler);
    signal(SIGABRT, Handler);
    signal(SIGPIPE, Handler);
    
    boost::mutex guard_mutex;
    
    // Setup readline
    history_filename = GetUserHomeDirectory() + "/.atomic_history";
    read_history(history_filename.data());
    
    rl_attempted_completion_function = AutoComplete;
    
    // Save command line state
    tcgetattr(fileno(stdin), &original_flags);
    
    // Parse commandline
    boost::program_options::options_description command_line;
    boost::program_options::variables_map variable_map;
    
    command_line.add_options()
    ("help,h",    "produce help message")
    ("command,c", boost::program_options::value<std::string>()->default_value(""), "command")
    ("server,s",  boost::program_options::value<std::string>()->default_value("localhost"), "server address")
    ("port,p",    boost::program_options::value<unsigned int>()->default_value(1202), "server port");
    
    try
    {
        boost::program_options::store(boost::program_options::command_line_parser(argc, argv).options(command_line).run(), variable_map);
    }
    catch (boost::program_options::unknown_option e)
    {
        std::cerr << e.what() << std::endl;
        std::cout << command_line << std::endl;
        CleanUp();
        return EXIT_FAILURE;
    }
    catch (boost::program_options::invalid_syntax e)
    {
        std::cerr << e.what() << std::endl;
        std::cout << command_line << std::endl;
        CleanUp();
        return EXIT_FAILURE;
    }
    
    if (variable_map.count("help") != 0)
    {
        std::cout << command_line << std::endl;
        CleanUp();
        return EXIT_SUCCESS;
    }
    
    net::Manager::Create();

    if (variable_map["command"].as<std::string>() == "")
    {
        std::cout << "\033[29;1mAtom Interactive Console, version " + std::string(VERSION) + " starting...\033[0m" << std::endl;
        std::cout << "\033[29;1mReleased under " + std::string(LICENSE) + ".\033[0m" << std::endl;
        std::cout << "Written by Mattias Runge 2010." << std::endl;
        
        std::cout << "Connecting to " << variable_map["server"].as<std::string>().data() << ":" << variable_map["port"].as<unsigned int>() << "..." << std::endl;
    }
    
    try
    {
        cc = ConsoleClient::Pointer(new ConsoleClient(variable_map["server"].as<std::string>(), variable_map["port"].as<unsigned int>()));
    }
    catch (std::runtime_error& e)
    {
        std::cerr << "Connection error: " << e.what() << std::endl;
        CleanUp();
        return EXIT_FAILURE;
    }
    
    if (variable_map["command"].as<std::string>() != "")
    {
        boost::mutex::scoped_lock guard(guard_mutex);
        on_message_condition.wait(guard);
        
        if (!finish)
        {
            cc->SendResponse(variable_map["command"].as<std::string>());
            
            if (!finish)
            {
                on_message_condition.wait(guard);
            }
        }
    }
    else
    {
        while (true)
        {
            boost::mutex::scoped_lock guard(guard_mutex);
            on_message_condition.wait(guard);
            
//.........这里部分代码省略.........

//.........这里部分代码省略.........
    configuration().addEntryPoint(EntryPoint(resource, path));
}

void WServer::removeEntryPoint(const std::string& path) {
    configuration().removeEntryPoint(path);
}

void WServer::restart(int argc, char **argv, char **envp)
{
#ifndef WT_WIN32
    char *path = realpath(argv[0], 0);

    // Try a few times since this may fail because we have an incomplete
    // binary...
    for (int i = 0; i < 5; ++i) {
        int result = execve(path, argv, envp);
        if (result != 0)
            sleep(1);
    }

    perror("execve");
#endif
}

void WServer::setCatchSignals(bool catchSignals)
{
    CatchSignals = catchSignals;
}

#if defined(WT_WIN32) && defined(WT_THREADED)

boost::mutex     terminationMutex;
bool             terminationRequested = false;
boost::condition terminationCondition;

void WServer::terminate()
{
    boost::mutex::scoped_lock terminationLock(terminationMutex);
    terminationRequested = true;
    terminationCondition.notify_all(); // should be just 1
}

BOOL WINAPI console_ctrl_handler(DWORD ctrl_type)
{
    switch (ctrl_type)
    {
    case CTRL_C_EVENT:
    case CTRL_BREAK_EVENT:
    case CTRL_CLOSE_EVENT:
    case CTRL_SHUTDOWN_EVENT:
    {
        WServer::terminate();
        return TRUE;
    }
    default:
        return FALSE;
    }
}

#endif

int WServer::waitForShutdown(const char *restartWatchFile)
{
#if !defined(WT_WIN32)
    if (!CatchSignals) {
        for(;;)

 void release_write_lock()
 {        
     boost::mutex::scoped_lock lock(mtx);
     is_current_writer = false;
     writer_finished.notify_all();
 }

 void f() {
     (*msg)();
     done = true;
     c.notify_one();
 }

void WServer::terminate()
{
    boost::mutex::scoped_lock terminationLock(terminationMutex);
    terminationRequested = true;
    terminationCondition.notify_all(); // should be just 1
}

int WServer::waitForShutdown(const char *restartWatchFile)
{
#if !defined(WT_WIN32)
    if (!CatchSignals) {
        for(;;)
            sleep(0x1<<16);
    }
#endif // WIN32

#ifdef WT_THREADED

#if !defined(WT_WIN32)
    sigset_t wait_mask;
    sigemptyset(&wait_mask);

    // Block the signals which interest us
    sigaddset(&wait_mask, SIGHUP);
    sigaddset(&wait_mask, SIGINT);
    sigaddset(&wait_mask, SIGQUIT);
    sigaddset(&wait_mask, SIGTERM);
    pthread_sigmask(SIG_BLOCK, &wait_mask, 0);

    for (;;) {
        int rc, sig= -1;

        // Wait for a signal to be raised
        rc= sigwait(&wait_mask, &sig);

        // branch based on return value of sigwait().
        switch (rc) {
        case 0: // rc indicates one of the blocked signals was raised.

            // branch based on the signal which was raised.
            switch(sig) {
            case SIGHUP: // SIGHUP means re-read the configuration.
                if (instance())
                    instance()->configuration().rereadConfiguration();
                break;

            default: // Any other blocked signal means time to quit.
                return sig;
            }

            break;
        case EINTR:
            // rc indicates an unblocked signal was raised, so we'll go
            // around again.

            break;
        default:
            // report the error and return an obviously illegitimate signal value.
            throw WServer::Exception(std::string("sigwait() error: ")
                                     + strerror(rc));
            return -1;
        }
    }

#else  // WIN32

    boost::mutex::scoped_lock terminationLock(terminationMutex);
    SetConsoleCtrlHandler(console_ctrl_handler, TRUE);
    while (!terminationRequested)
        terminationCondition.wait(terminationLock);
    SetConsoleCtrlHandler(console_ctrl_handler, FALSE);
    return 0;

#endif // WIN32
#else
    return 0;
#endif // WT_THREADED
}

  /// Entry point for the asynchronous monitor thread
  void run()
  {
    bool run = true;
    typedef std::pair< callback_t , std::pair< file_action, std::string> > notification;
    std::vector< notification> notifications;

    for (;;)
    {
      {
        // Lock and check m_run; if it was set to false, we must return ASAP
        boost::mutex::scoped_lock lock(m_run_mutex);
        run = m_run;
        if (!run)
          return;
        // We release the lock, block the thread until one second
        m_run_cond.timed_wait(lock, moost::thread::xtime_util::add_ms(moost::thread::xtime_util::now(), m_sleep_ms));
      }
      // If we are not running (e.g. when the destructor woke us up), return
      if (!run)
        return;

      // Clear the notifications vector where we will collect the events
      // that will be fired
      notifications.clear();
      {
        // Lock m_file_mutex while we are working on m_file_callback
        boost::mutex::scoped_lock lock(m_file_mutex);
        for (std::map<std::string, callback_t>::iterator it = m_file_callback.begin(); it != m_file_callback.end(); ++it)
        {
          boost::filesystem::path p(it->first);

          // Does the path exist?
          std::time_t lw = last_write_time(p);
          if (lw != 0)
          {
            // Check its last modification time and compare it with what we had earlier
            std::map< std::string, std::pair<time_t, time_t> >::iterator it_mod = m_file_modified.find(it->first);

            if (it_mod == m_file_modified.end())
            {
              // We haven't seen this file so far, so insert it into the
              // map and add a creation event that will be fired
              m_file_modified[it->first] = std::make_pair(lw, lw);
              notifications.push_back(std::make_pair(it->second, std::make_pair(CREATED, it->first)));
            }
            else
            {
              // only case we consider a real modification: prev prev mod != prev mod,
              // but this mod == prev mod
              // the idea is that we want to capture a write to a file,
              // but only notify when the write is finished

              /**
               * \todo This could cause problems with frequent writing.
               *       We should really use boost.interprocess file locking
               *       instead (when we get 1.36 everywhere)
               */
              if (lw == it_mod->second.second && it_mod->second.first != it_mod->second.second)
                notifications.push_back(std::make_pair(it->second, std::make_pair(CHANGED, it->first)));
              it_mod->second.first = it_mod->second.second;
              it_mod->second.second = lw;
            }
          }
          else
          {
            // The path does not exist. Did we have it before? If so, fire
            // a deletion event.
            std::map< std::string, std::pair<time_t, time_t> >::iterator it_mod = m_file_modified.find(it->first);
            if (it_mod != m_file_modified.end())
            {
              m_file_modified.erase(it_mod);
              notifications.push_back(std::make_pair(it->second, std::make_pair(DELETED, it->first)));
            }
          }
        }
      }

      // okay!  we've released our lock on m_file_callback and m_file_modified
      // so it's time to send off our notifications
      for (std::vector<notification>::iterator it = notifications.begin(); it != notifications.end(); ++it)
      {
        try
        {
          it->first(it->second.first, it->second.second);
        }
        catch (...)
        {
          // \todo  can we do better here than silently ignoring the exception?
        }
      }
    }
  }

  void sort_()
  {
    uint64_t count = 0;
    while (count< count_)
    {
      uint32_t pre_buf_size = 0;
      uint32_t pre_buf_num = 0;
      {
        boost::mutex::scoped_lock lock(pre_buf_mtx_);
        while (pre_buf_size_==0)
          pre_buf_con_.wait(lock);

        assert(pre_buf_size_ <= RUN_BUF_SIZE_);
        memcpy(run_buf_, pre_buf_, pre_buf_size_);
        pre_buf_size = pre_buf_size_;
        pre_buf_num = pre_buf_num_;
        count += pre_buf_num_;
        pre_buf_num_ = pre_buf_size_ = 0;
        pre_buf_con_.notify_one();
      }

      key_buf_ = (struct KEY_PTR*)realloc(key_buf_, pre_buf_num*sizeof(struct KEY_PTR));

      uint32_t pos = 0;
      for (uint32_t i = 0; i<pre_buf_num; ++i)
      {
        key_buf_[i] = KEY_PTR(pos);
        assert(pos <= RUN_BUF_SIZE_);
        pos += *(LEN_TYPE*)(run_buf_ + pos)+sizeof(LEN_TYPE);
        IASSERT(pos<=pre_buf_size);
      }
      IASSERT(pos==pre_buf_size);

      quick_sort_(0, pre_buf_num-1, pre_buf_num);

      boost::mutex::scoped_lock lock(out_buf_mtx_);
      while (out_buf_size_ != 0)
        out_buf_con_.wait(lock);

      out_buf_size_ = 0;
      out_buf_num_ = 0;
      LEN_TYPE max_len_of_this_run = (LEN_TYPE)0;
      for (uint32_t i=0; i<pre_buf_num; ++i, ++out_buf_num_)
      {
        assert(key_buf_[i].pos <= RUN_BUF_SIZE_);
        assert(out_buf_size_+key_buf_[i].LEN(run_buf_)+ sizeof(LEN_TYPE) <= RUN_BUF_SIZE_);
        memcpy(out_buf_+out_buf_size_, run_buf_+ key_buf_[i].pos, key_buf_[i].LEN(run_buf_)+ sizeof(LEN_TYPE));
        LEN_TYPE len = key_buf_[i].LEN(run_buf_) + sizeof(LEN_TYPE);
        out_buf_size_ += len;
        if(len > max_len_of_this_run) max_len_of_this_run = len;
      }
      max_record_len_of_this_run_ = max_len_of_this_run;
      min_run_buff_size_for_merger_ += max_record_len_of_this_run_;
      if(max_len_of_this_run > max_record_len_) max_record_len_ = (uint32_t)max_len_of_this_run;

      IASSERT(out_buf_num_ == pre_buf_num);
      IASSERT(out_buf_size_ == pre_buf_size);

      out_buf_con_.notify_one();
    }

    std::cout<<"Sorting is over...\n";
  }

  void prefetch_(FILE* f)
  {
    IO_TYPE ioStream(f);
    const uint64_t FILE_LEN = ioStream.length();
    run_num_ = 0;
    uint64_t pos = sizeof(uint64_t);
    std::cout<<std::endl;
    ioStream.seek(pos);

    while(pos < FILE_LEN)
    {
      std::cout<<"\rA runner is processing "<<pos*1./FILE_LEN<<std::flush;
      ++run_num_;
      boost::mutex::scoped_lock lock(pre_buf_mtx_);

      while (pre_buf_size_!=0)
        pre_buf_con_.wait(lock);

      //uint32_t s = (uint32_t)(FILE_LEN-pos>RUN_BUF_SIZE_? RUN_BUF_SIZE_: FILE_LEN-pos);
      uint32_t s;
      if(!ioStream.isCompression())
         s = (uint32_t)(FILE_LEN-pos>RUN_BUF_SIZE_? RUN_BUF_SIZE_: FILE_LEN-pos);
      else
         s = RUN_BUF_SIZE_;
      //std::cout<<std::endl<<pos<<"-"<<FILE_LEN<<"-"<<RUN_BUF_SIZE_<<"-"<<s<<std::endl;
      if(!ioStream.isCompression())
        ioStream.seek(pos);
      //IASSERT(fread(pre_buf_, s, 1, f)==1);
      s = ioStream.read(pre_buf_, s);
      if(!ioStream.isCompression())
        pos += (uint64_t)s;
      else
        pos = ioStream.tell();

      //check the position of the last record
      pre_buf_size_ = 0;
      pre_buf_num_ = 0;
      for(; pre_buf_size_<s; ++pre_buf_num_)
      {
        if (pre_buf_size_+*(LEN_TYPE*)(pre_buf_+pre_buf_size_)+sizeof(LEN_TYPE)>s)
          break;
        pre_buf_size_ += *(LEN_TYPE*)(pre_buf_+pre_buf_size_)+sizeof(LEN_TYPE);
      }
      pos -= (uint64_t)(s- pre_buf_size_);

      //std::cout<<"pre_buf_size_ "<<pre_buf_size_<<" pre_buf_num_ "<<pre_buf_num_<<" ret "<<s<<" pos "<<pos<<std::endl;

      if (pre_buf_num_ == 0)
      {
        std::cout<<"\n[Warning]: A record is too long, and has been ignored!\n";
        //pos += *(LEN_TYPE*)(pre_buf_+pre_buf_size_) + sizeof(LEN_TYPE);
        --count_;
        RUN_BUF_SIZE_ = (uint32_t)((*(LEN_TYPE*)(pre_buf_+pre_buf_size_) + sizeof(LEN_TYPE))*1.1);
        pre_buf_ = (char*)realloc(pre_buf_, RUN_BUF_SIZE_);
        continue;
      }

      //IASSERT(pre_buf_size_ <= RUN_BUF_SIZE_);
      pre_buf_con_.notify_one();
    }
    std::cout<<"Prefetching is over...\n";
  }

inline void
boost_threaded_monitor::notify_all() {
	cond_.notify_all();
}