void A::RunThreads() {
    myProducerThread = boost::thread(&A::RunProducer, this);
    myConsumerThread = boost::thread(&A::RunConsumer, this);

    myProducerThread.join();
    myConsumerThread.join();
}

inline connection::~connection()
{
    socket_m.close();
    
    read_thread_m.join();
    send_thread_m.join();
}

 virtual ~Log()
 {
     _isStopping = true;
     _threadPool.interrupt_all();
     _threadPool.join_all();
     _stringLoggerThread.interrupt();
     _stringLoggerThread.join();
 }

    //! デバイスを開く
    //! 開くデバイスの指定は、現在WAVE_MAPPER固定。
    //! 簡単のため例外安全性などはあまり考慮されていない点に注意。
	bool OpenDevice(size_t sampling_rate, size_t channel, size_t block_size, size_t multiplicity, callback_function_t callback)
	{
		BOOST_ASSERT(0 < block_size);
		BOOST_ASSERT(0 < multiplicity);
		BOOST_ASSERT(0 < channel && channel <= 2);
		BOOST_ASSERT(callback);
		BOOST_ASSERT(!process_thread_.joinable());

		block_size_ = block_size;
		channel_ = channel;
		callback_ = callback;
		multiplicity_ = multiplicity;

        //! デバイスがオープン完了するまでcallbackが呼ばれないようにするためのロック
		boost::unique_lock<boost::mutex> lock(initial_lock_mutex_);

		terminated_ = false;
		process_thread_ = boost::thread([this] { ProcessThread(); });

		WAVEFORMATEX wf;
		wf.wFormatTag = WAVE_FORMAT_PCM;
		wf.nChannels = 2;
		wf.wBitsPerSample = 16;
		wf.nBlockAlign = wf.nChannels * (wf.wBitsPerSample / 8);
		wf.nSamplesPerSec = sampling_rate;
		wf.nAvgBytesPerSec = wf.nBlockAlign * wf.nSamplesPerSec;
		wf.cbSize = sizeof(WAVEFORMATEX);

		headers_.resize(multiplicity_);
		for(auto &header: headers_) {
			header.reset(new WaveHeader(block_size * channel * sizeof(short)));
		}

        //! WAVEHDR使用済み通知を受け取る方式として
        //! CALLBACK_FUNCTIONを指定。
        //! 正常にデバイスがオープンできると、
        //! waveOutWriteを呼び出した後でWaveOutProcessor::waveOutProcに通知が来るようになる。
		MMRESULT const result = 
			waveOutOpen(
				&hwo_, 
				0, 
				&wf, 
				reinterpret_cast<DWORD>(&WaveOutProcessor::waveOutProc), 
				reinterpret_cast<DWORD_PTR>(this),
				CALLBACK_FUNCTION
				);

		if(result != MMSYSERR_NOERROR) {
			terminated_ = true;
			process_thread_.join();
			terminated_ = false;
            hwo_ = NULL;

			return false;
		}

		return true;
	}

  virtual void killTask(ExecutorDriver* driver, const TaskID& taskId) {
                  if (thread) {
                    thread->interrupt();
                    thread->join();
                    delete thread;
                    thread = 0;
                  }
	  	  driver->sendFrameworkMessage("Executor " + host_name+ " KILLING TASK");
		  driver->stop();
}

  virtual void shutdown(ExecutorDriver* driver) {
  	driver->sendFrameworkMessage("Executor " + host_name+ "SHUTTING DOWN");
                  if (thread) {
                    thread->interrupt();
                    thread->join();
                    delete thread;
                    thread = 0;
                  }
	driver->stop();
  }

void cmd_connect_to(string connect_to)
{
    auto devlist = LibusbInterface::listDevices(0xF539, 0xF539);
    int i = 0;

    if(devlist.empty())
    {
        cout << "    No devices found!" << endl;
        return;
    }

    if(devlist.size() > 1)
    {
        if(connect_to.compare("") == 0)
        {
            cout << "    Select a device" << endl;
            for(auto dev : devlist)
            {
                cout << "        " << dev.first << "\t" << dev.second << endl;
            }
            return;
        }
        else
        {
            chosen_serial = connect_to;
        }
    }
    else
    {
        chosen_serial = devlist[0].first;
    }

    if(connected)
    {
        liObj->endSignal();
        dpObj->endSignal();
        connected = false;
        thread1.join();
        thread2.join();
        delete liObj;
        delete dpObj;
        dpObj = NULL;
    }

    dQueue.reset(new queue<DataSet>());
    liObj = new LibusbInterface(&bmutex, dQueue.get(), 0xF539, 0xF539, chosen_serial);
    dpObj = new DataProcessor(&bmutex, dQueue.get());

    thread1 = boost::thread(bind(&LibusbInterface::operator(),liObj));  // Bind prevents copying obj (need to keep ptr)
    thread2 = boost::thread(bind(&DataProcessor::operator(),dpObj));

    connected = true;
    cout << "    Connected to device, serial: " << chosen_serial << endl;
}

void shutdown()
{
  boost::recursive_mutex::scoped_lock lock(g_shutting_down_mutex);
  // gracefully return if we've not fired ros up yet, or mid-shutdown
  if (g_shutting_down || !g_initialized)
  {
    return;
  }

  ROSCPP_LOG_DEBUG("Shutting down roscpp");

  g_shutting_down = true;

  g_global_queue->disable();
  g_global_queue->clear();

  if (g_internal_queue_thread.get_id() != boost::this_thread::get_id())
  {
    g_internal_queue_thread.join();
  }

  const log4cxx::LoggerPtr& logger = log4cxx::Logger::getLogger(ROSCONSOLE_ROOT_LOGGER_NAME);
  logger->removeAppender(g_rosout_appender);
  g_rosout_appender = 0;

  // reset this so that the logger doesn't get crashily destroyed
  // again during global destruction.  
  //
  // See https://code.ros.org/trac/ros/ticket/3271
  //
  log4cxx::Logger::getRootLogger()->getLoggerRepository()->shutdown();
  log4cxx::LoggerPtr& fo_logger = ros::file_log::getFileOnlyLogger();
  fo_logger = log4cxx::LoggerPtr();
  
  if (g_started)
  {
    TopicManager::instance()->shutdown();
    ServiceManager::instance()->shutdown();
    PollManager::instance()->shutdown();
    ConnectionManager::instance()->shutdown();
    XMLRPCManager::instance()->shutdown();
  }

  WallTime end = WallTime::now();

  ROSCPP_LOG_DEBUG("Shutdown finished");

  g_started = false;
  g_ok = false;
  Time::shutdown();
}

void cmd_exit()
{
    cout << "Exiting...\n";
    if(connected)
    {
        liObj->endSignal();
        dpObj->endSignal();
        connected = false;
        thread1.join();
        thread2.join();
        delete liObj;
        delete dpObj;
    }
    running = false;
}

inline threadTest::~threadTest()
{
  _glInit = false;

  _glThread.interrupt();
  std::cout << "Interrupting main thread" << std::endl;
  fflush(stdout);

  _glThread.join();
  std::cout << "Joinning main thread" << std::endl;
  fflush(stdout);

  printTest();
  std::cout << "Main thread destroyed" << std::endl;
  fflush(stdout);
};

		/*
		 *	beginLawnmowerThread
		 *		Starts the lawnmower code.
		 */
		bool beginLawnmowerThread() {
			if ( thread_1.timed_join(boost::posix_time::milliseconds(10)) ) {
				cout << "\033[36m[THRIFT]\033[0m Beginning lawnmower thread." << endl;
				
				if (waypoints_list.size() != 2) {
					cout << "\033[31m[THRIFT]\033[0m Cannot start lawnmower without boundaries!" << endl;
					return false;
				}
				
				Pos corner1;
				Pos corner2;
				
				corner1.lat = waypoints_list.front().lat;
				corner1.lon = waypoints_list.front().lon;
				corner2.lat = waypoints_list.back().lat;
				corner2.lon = waypoints_list.back().lon;
				
				usingWindows = false;
				exitProgram = false;
				userState = -1;
				thread_1 = boost::thread(run_lawnmower, boost::ref(fb), boost::ref(gps), boost::ref(imu), boost::ref(buzzer), corner1, corner2);
				return true;
			} else {
				cout << "\033[31m[THRIFT]\033[0m Cannot start lawnmower, copter is flying." << endl;
				return false;
			}
		}

    //! デバイスを閉じる
	void CloseDevice() {
		terminated_.store(true);
		process_thread_.join();
		waveOutReset(hwo_);
		waveOutClose(hwo_);

        //! waveOutResetを呼び出したあとで
        //! WAVEHDRの使用済み通知が来ることがある。
        //! つまり、waveOutResetを呼び出した直後に
        //! すぐにWAVEHDRを解放できない(デバイスによって使用中かもしれないため)
        //! そのため、確実にすべてのWAVEHDRの解放を確認する。
		for( ; ; ) {
			int left = 0;
			for(auto &header : headers_ | boost::adaptors::indirected) {
				if(header.get()->dwUser == WaveHeader::DONE) {
					waveOutUnprepareHeader(hwo_, header.get(), sizeof(WAVEHDR));
					header.get()->dwUser = WaveHeader::UNUSED;
				} else if(header.get()->dwUser == WaveHeader::USING) {
					left++;
				}
			}

			if(!left) { break; }
			Sleep(10);
		}
		hwo_ = NULL;
	}

 void test_echo_service()
 {
   service_class_thread = nullptr;
   // Create a server on port 12345
   fr::socket::socket_server<fr::socket::server_body<echo_service> > my_echo_server(12345);
   my_echo_server.thread_spawned.connect(boost::bind(&test_client_server::set_service_class_thread, this, _1));
   boost::thread *server_thread = my_echo_server.start();
   // Give the server time to start
   boost::this_thread::yield();
   fr::socket::socket_client echo_client("localhost", 12345);
   std::string expected("Hello, sockets!");
   std::string actual;
   echo_client.streams->stream_out << expected << std::endl;
   getline(echo_client.streams->stream_in, actual);
   //    std::cout << std::endl << "Expected: \"" << expected << "\"" << std::endl;
   //    std::cout << "Actual: \"" << actual << "\"" << std::endl;
   CPPUNIT_ASSERT(expected == actual);
   my_echo_server.shutdown();
   // Poke the echo server so it exits. IRL using blocking reads in
   // your server code opens a sack of pain. For this test, it was
   // a significantly smaller sack of pain than having to deal with
   // select and signals or polling the stream and building the
   // string up that way. No matter how you go about it, there's always
   // a good bit of suck associated with threaded I/O.
   echo_client.streams->stream_out << std::endl;
   if (nullptr == service_class_thread) {
     CPPUNIT_FAIL("Service class thread never signaled");
   }
   service_class_thread->join();
   delete service_class_thread;
   server_thread->join();
 }

  // Destructor : stop the brackground thread first !
  inline ~pyPTAM() {
    if (is_slam_started) {
      s->Stop();
      is_slam_started = false;
    }

    sys_thread->join();
  }

    ~app_state_t() {
        COCAINE_LOG_TRACE(log, "application is destroying its internal state");

        work.reset();
        thread.join();

        COCAINE_LOG_TRACE(log, "application has destroyed its internal state");
    }

 ~AsyncCRC32()
 {
   mutex.lock();
   finished = true;
   mutex.unlock();
   
   readCond.notify_one();
   thread.join();
   for (auto buf : queue) delete buf;
 }

int Worker::Abort()
{
    // lock nothing coz care nothing ...

    if(m_state <= 0) return m_state;
    m_state = -1;

    if(m_thread) m_thread->interrupt(); // force to get out

    //usleep(50 * 1000);
    return m_state;
}

void LoopbackControllerManager::fini()
{
  ROS_DEBUG("calling LoopbackControllerManager::fini");

  //pr2_hardware_interface::ActuatorMap::const_iterator it;
  //for (it = hw_.actuators_.begin(); it != hw_.actuators_.end(); ++it)
  //  delete it->second; // why is this causing double free corrpution?
  cm_->~ControllerManager();
  rosnode_->shutdown();

  ros_spinner_thread_->join();
}

int Worker::Stop()
{
    // lock nothing coz care nothing ...

    if(m_state <= 0) return m_state;
    m_state = 0;

    m_condition.notify_one(); // just send out a signal

    if(m_thread) m_thread->join(); // and wait for end

    m_state = -1;
    return m_state;
}

void nuke_ms::clientnode::catchThread(boost::thread& thread, unsigned threadwait_ms)
{
    // a thread id that compares equal to "not-a-thread"
    boost::thread::id not_a_thread;

    try {
        // give the thread a few seconds time to join
        thread.timed_join(boost::posix_time::millisec(threadwait_ms));
    }
    catch(...)
    {}

    // if the thread finished, return. otherwise try to kill the thread
    if (thread.get_id() == not_a_thread)
        return;

    thread.interrupt();

    // if it is still running, let it go
    if (thread.get_id() == not_a_thread)
        return;

    thread.detach();
}