<commit_before>#ifndef _MSC_VER
# include <unistd.h>
#else
# include <io.h>
#endif
#include <fcntl.h>

#include "libport/thread.hh"
#include "libport/assert.hh"

#include "urbi/usyncclient.hh"
#include "urbi/uconversion.hh"

/* "min" shouldn't be defined as a preprocessor macro. On windows, it is. So
 * using std::min leads to a parse error because CPP expands it as
 *   std::((a) < (b) ? (a) : (b))
 * However, this should only occur on windows with the STL of MS VC++.
 */
#ifdef min
# ifdef WIN32
#  undef min
# else
#  error "min is defined as a CPP macro and we're not on WIN32. Report this!"
# endif /* !WIN32 */
#endif /* !min */

namespace urbi
{
  USyncClient::USyncClient(const char *_host, int _port, int _buflen)
    : UClient(_host, _port, _buflen),
      sem_(),
      queueLock_(),
      msg(0),
      syncLock_(),
      syncTag (""),
      stopCallbackThread_(false)
  {
    if (error())
      return;
    libport::startThread(this, &USyncClient::callbackThread);
    if (!defaultClient)
      defaultClient = this;
  }

  void USyncClient::callbackThread()
  {
    while (true)
    {
      sem_--;
      if (stopCallbackThread_)
	return;
      queueLock_.lock();
      if (queue.empty())
      {
	//we got mysteriously interrupted
	queueLock_.unlock();
	continue;
      }
      UMessage *m = queue.front();
      queue.pop_front();
      queueLock_.unlock();
      UAbstractClient::notifyCallbacks(*m);
      delete m;
    }
  }
  void USyncClient::stopCallbackThread()
  {
    stopCallbackThread_ = true;
  }
  void USyncClient::processEvents(const libport::utime_t timeout)
  {
    libport::utime_t startTime = libport::utime();
    while (timeout < 0 || libport::utime() - startTime <= timeout)
    {
      queueLock_.lock();
      if (queue.empty())
      {
	queueLock_.unlock();
	return;
      }
      UMessage *m = queue.front();
      queue.pop_front();
      queueLock_.unlock();
      UAbstractClient::notifyCallbacks(*m);
      delete m;
    }

  }

  void USyncClient::notifyCallbacks(const UMessage &msg)
  {
    queueLock_.lock();
    if (!syncTag.empty() && syncTag == msg.tag)
    {
      this->msg = new UMessage(msg);
      syncLock_++;
      syncTag = "";
    }
    else
    {
      queue.push_back(new UMessage(msg));
      sem_++;
    }
    queueLock_.unlock();
  }

  UMessage* USyncClient::waitForTag(const char* tag)
  {
    syncTag = tag;
    queueLock_.unlock();
    syncLock_--;
    syncTag = "";
    return msg;
  }

  UMessage* USyncClient::syncGet_(const char* format,
				  const char* mtag, const char* mmod,
				  va_list& arg)<commit_msg>Remove a race condition in USyncClient, that could trigger in syncGet*.<commit_after>#ifndef _MSC_VER
# include <unistd.h>
#else
# include <io.h>
#endif
#include <fcntl.h>

#include "libport/thread.hh"
#include "libport/assert.hh"

#include "urbi/usyncclient.hh"
#include "urbi/uconversion.hh"

/* "min" shouldn't be defined as a preprocessor macro. On windows, it is. So
 * using std::min leads to a parse error because CPP expands it as
 *   std::((a) < (b) ? (a) : (b))
 * However, this should only occur on windows with the STL of MS VC++.
 */
#ifdef min
# ifdef WIN32
#  undef min
# else
#  error "min is defined as a CPP macro and we're not on WIN32. Report this!"
# endif /* !WIN32 */
#endif /* !min */

namespace urbi
{
  USyncClient::USyncClient(const char *_host, int _port, int _buflen)
    : UClient(_host, _port, _buflen),
      sem_(),
      queueLock_(),
      msg(0),
      syncLock_(),
      syncTag (""),
      stopCallbackThread_(false)
  {
    if (error())
      return;
    libport::startThread(this, &USyncClient::callbackThread);
    if (!defaultClient)
      defaultClient = this;
  }

  void USyncClient::callbackThread()
  {
    while (true)
    {
      sem_--;
      if (stopCallbackThread_)
	return;
      queueLock_.lock();
      if (queue.empty())
      {
	//we got mysteriously interrupted
	queueLock_.unlock();
	continue;
      }
      UMessage *m = queue.front();
      queue.pop_front();
      queueLock_.unlock();
      UAbstractClient::notifyCallbacks(*m);
      delete m;
    }
  }
  void USyncClient::stopCallbackThread()
  {
    stopCallbackThread_ = true;
  }
  void USyncClient::processEvents(const libport::utime_t timeout)
  {
    libport::utime_t startTime = libport::utime();
    while (timeout < 0 || libport::utime() - startTime <= timeout)
    {
      queueLock_.lock();
      if (queue.empty())
      {
	queueLock_.unlock();
	return;
      }
      UMessage *m = queue.front();
      queue.pop_front();
      queueLock_.unlock();
      UAbstractClient::notifyCallbacks(*m);
      delete m;
    }

  }

  void USyncClient::notifyCallbacks(const UMessage &msg)
  {
    queueLock_.lock();
    if (!syncTag.empty() && syncTag == msg.tag)
    {
      this->msg = new UMessage(msg);
      syncTag.clear();
      syncLock_++;
    }
    else
    {
      queue.push_back(new UMessage(msg));
      sem_++;
    }
    queueLock_.unlock();
  }

  UMessage* USyncClient::waitForTag(const char* tag)
  {
    syncTag = tag;
    queueLock_.unlock();
    syncLock_--;
    //syncTag is reset by the other thread
    return msg;
  }

  UMessage* USyncClient::syncGet_(const char* format,
				  const char* mtag, const char* mmod,
				  va_list& arg)
{'fraction_non_alphanumeric': 0.13838268792710706, 'fraction_numerical': 0.0030372057706909645, 'mean_word_length': 2.7716535433070866, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7134255', 'n_tokens_mistral': 1822, 'n_tokens_neox': 1766, 'n_words': 435}