[Boost] Threads in eine Reihenfolge zwingen

[Dravere]

Grüsse zusammen,

Ich steh glaub grad etwas auf dem Schlauch und gehe deswegen gleich noch etwas Ablenkung suchen, aber um vielleicht noch ein paar zusätzliche Ideen zu bekommen, wollte ich mal hier nach einer Lösung fragen.

Mein Problem ist eigentlich relativ einfach. Ich habe n Threads, welche alle etwas von ein und demselben Http-Server haben wollen, allerdings unterschiedliche Daten. Es gibt daher genau eine TCP-Verbindung. Um das ganze zu beschleunigen, sollen HTTP-Request abgeschickt werden, bevor überhaupt die erste HTTP-Response zurückkommt. Der Ping zum Server ist >200ms, daher kann ich da durchaus Zeit sparen.

Jetzt ist aber das Problem vorhanden, dass wenn Thread x ein HTTP-Request verschickt, und dann Thread x+1 kommt und ein HTTP-Request verschickt, dann Thread x+2 usw., muss gewährleistet sein, dass in der genau gleichen Reihenfolge die Threads die HTTP-Request empfangen.

Vielleicht noch etwas Pseudo-Code:

void foo()
{
  // Synchronisiere die Send-Funktion.
  send();
  // Beende die Synchronisierung der Send-Funktion.

  // Die Send-Funktion muss immer aufrufbar sein, solange nicht ein anderer Thread gerade am senden ist.
  // Stelle nun sicher, dass die Reihenfolge der Threads bei der Send- und Receive-Funktion dieselbe ist.

  // Syncronisiere die Receive-Funktion.
  receive();
  // Beende die Synchronisierung der Receive-Funktion.
}

Ich hoffe ihr könnt mir etwas auf die Sprünge helfen.

Grüssli

[tipper]

Das ist Aufgabe der verwendeten HTTP-Lib, wenn du diese selber schreibst, so musst du für die Multi-Thread Sicherheit selber sorgen.
Davon einfach in jedem Socket die HTTP-Requests in Form von Strings zu versenden rate ich dir ab, dann hast du nämlich genau dein Problem.

[Badestrand]

Dravere schrieb:

Jetzt ist aber das Problem vorhanden, dass wenn Thread x ein HTTP-Request verschickt, und dann Thread x+1 kommt und ein HTTP-Request verschickt, dann Thread x+2 usw., muss gewährleistet sein, dass in der genau gleichen Reihenfolge die Threads die HTTP-Request empfangen.

Meinst du bei dem von mir fett unterlegten Teil statt "Request" "Antwort"? Wenn ja: Kann zwar sein, dass ich vollkommen danebenschieße, aber evtl tut's ein deque mit Thread-IDs (oder so); Threads die 'nen Request gesendet haben, tragen sich hinten ein und nur der Thread, der vorne steht, darf die aktuelle Antwort bearbeiten (und dann natürlich pop_front).

[Dravere]

@tipper,
Wenn das die HTTP-Bibliothek übernimmt, dann hat man meistens das Problem, dass nicht mehrere HTTP-Requests gleichzeitig verschickt werden können. Diese HTTP-Lib, welche ich verwende, ist sehr low-level.

@Badestrand,
Jap, Schreibfehler, ist HTTP-Response gemeint.

Bisher gefundene Lösung:

class HttpConnection
{
  // Inner class //
private:
  class Requester
  {
    // Attributes //
  private:
    dsi::net::http::HttpResponse* m_response;

    boost::mutex m_waitMutex;
    boost::unique_lock<boost::mutex> m_waitLock;

    // Constructors & Destructor //
  public:
    Requester(dsi::net::http::HttpResponse* response)
      , m_response(response)
      , m_waitMutex()
      , m_waitLock(m_waitMutex)
    {
    }

    ~Requester() { };

    // Methods //
  public:
    wait() { boost::lock_guard<boost::mutex> sync(m_waitMutex); };
    stop_wait() { m_waitLock.unlock(); };

    dsi::net::http::HttpResponse& get_response() { return *m_response; };
  };

  // Typedef //
private:
  typedef std::list<Requester*> RequesterList_t;

  // Attributes //
private:
  std::string m_server;
  dsi::net::http::Http m_http;
  RequesterList_t m_requesterList;

  boost::thread* m_thread;
  boost::mutex m_listMutex;
  boost::mutex m_requestMutex;
  boost::condition_variable m_waitCondition;

  // Constructors & Destructor //
public:
  HttpConnection(std::string const& server)
    : m_server(server)
    , m_http()
    , m_requesterList()
    , m_thread(nullptr)
    , m_listMutex()
    , m_requestMutex()
    , m_waitCondition()
  {
  }

  ~HttpConnection()
  {
    stop();
  }

  // Methods //
private:
  void thread_run()
  {
    Requester* requester;

    for(;;)
    {
      {
        boost::unique_lock<boost::mutex> sync(m_listMutex);

        while(m_requesterList.empty())
        { m_waitCondition.wait(sync); }

        requester = m_requestList.front();
        m_requestList.pop_front();
      }

      if(!m_http.is_connected() && !m_http.connect(m_server))
      { /* error report */ }

      m_http.receive_http_response(requester->get_response());

      requester->unlock();
    }
  }

public:
  void start()
  {
    if(!m_thread)
    { m_thread = new boost::thread(boost::bind(&HttpConnection::thread_run, this)); }
  }

  void stop()
  {
    if(m_thread)
    {
      m_thread->interrupt();
      delete m_thread;
      m_thread = nullptr;
    }
  }

  void request_data(dsi::net::http::HttpRequest& request, dsi::net::http::HttpResponse& response)
  {
    Requester* requester = new Requester(&response);

    {
      boost::lock_guard<boost::mutex> listSync(m_listMutex);
      boost::lock_guard<boost::mutex> requestSync(m_requestMutex);

      m_http.send_http_message(request);

      m_requesterList.push_back(requester);

      m_waitCondition.notify_one();
    }

    requester->wait();

    delete requester;
  }
};

Hab es noch nicht getestet, werde das erste heute Abend machen. So ganz will es mir aber noch nicht gefallen.

Grüssli