src/commons/st/st.h

#ifndef COMMONS_ST_ST_H
#define COMMONS_ST_ST_H

#include <algorithm>
#include <boost/foreach.hpp>
#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <commons/array.h>
#include <commons/delegates.h>
#include <commons/nullptr.h>
#include <commons/sockets.h>
#include <commons/utility.h>
#include <exception>
#include <map>
#include <queue>
#include <set>
#include <sstream>
#include <st.h>
#include <stx.h>

#define foreach BOOST_FOREACH
#define shared_ptr boost::shared_ptr

namespace commons
{
  using namespace boost;
  using namespace std;

  enum { default_stack_size = 65536 };

  struct stfd_closer {
    static void apply(st_netfd_t fd) { check0x(st_netfd_close(fd)); }
  };

  typedef closing<st_netfd_t, stfd_closer> st_closing;

  class st_lock
  {
    NONCOPYABLE(st_lock)
    public:
      st_lock(st_mutex_t mx) : mx_(mx) { check0x(st_mutex_lock(mx)); }
      ~st_lock() { check0x(st_mutex_unlock(mx_)); }
    private:
      st_mutex_t mx_;
  };

  st_thread_t
  st_spawn(const fn& f)
  {
    return st_thread_create(&run_function0_null,
                            new fn(f),
                            true,
                            default_stack_size);
  }

  void
  st_join(st_thread_t t)
  {
    check0x(st_thread_join(t, nullptr));
  }

  st_netfd_t
  st_tcp_connect(in_addr host, uint16_t port, st_utime_t timeout)
  {
    // Create remote socket address.
    sockaddr_in sa = make_sockaddr(host, port);

    // Create the socket.
    st_closing s(checkpass(st_netfd_open_socket(tcp_socket(true))));

    // Connect.
    check0x(st_connect(s.get(), reinterpret_cast<sockaddr*>(&sa), sizeof sa, timeout));
    return s.release();
  }

  st_netfd_t
  st_tcp_connect(const char *host, uint16_t port, st_utime_t timeout)
  {
    in_addr ipaddr;

    // First try to parse as IP address.  Note: inet_addr() is obsolete.  Note:
    // inet_aton returns 0 if address is invalid.
    if (inet_aton(host, &ipaddr) == 0) {
      // Then look up by hostname.
      check0x(stx_dns_getaddr(host, &ipaddr, timeout));
    }

    return st_tcp_connect(ipaddr, port, timeout);
  }

  st_netfd_t
  st_tcp_listen(uint16_t port)
  {
    int sfd = tcp_listen(port);
    try {
      // Create a net file descriptor around a listener socket.
      return checkpass(st_netfd_open_socket(sfd));
    } catch (...) {
      close(sfd);
      throw;
    }
  }

  class st_cond
  {
    NONCOPYABLE(st_cond)
    public:
      st_cond() : c(checkerr(st_cond_new())) {}
      ~st_cond() { check0x(st_cond_destroy(c)); }
      void wait() { check0x(st_cond_wait(c)); }
      void wait(st_utime_t t) { check0x(st_cond_timedwait(c, t)); }
      void signal() { st_cond_signal(c); }
      void bcast() { st_cond_broadcast(c); }
    private:
      st_cond_t c;
  };

  class st_bool
  {
    public:
      st_bool(bool init = false) : c(), b(init) {}
      void set() { b = true; c.bcast(); }
      void reset() { b = false; c.bcast(); }
      void waitset() { if (!b) c.wait(); }
      void waitreset() { if (b) c.wait(); }
      operator bool() { return b; }
    private:
      st_cond c;
      bool b;
  };

  void toggle(st_bool& b) { if (b) b.reset(); else b.set(); }

  class st_mutex
  {
    NONCOPYABLE(st_mutex)
    public:
      st_mutex() : m(checkerr(st_mutex_new())) {}
      ~st_mutex() { check0x(st_mutex_destroy(m)); }
      void lock() { check0x(st_mutex_lock(m)); }
      bool trylock() {
        int res = st_mutex_trylock(m);
        if (res == 0) return true;
        else if (errno == EBUSY) return false;
        else check0x(res);
      }
      void unlock() { check0x(st_mutex_unlock(m)); }
    private:
      st_mutex_t m;
  };

  template <typename T>
  class st_channel
  {
    public:
      void push(const T &x) {
        q_.push(x);
        empty_.signal();
      }
      T take() {
        while (q_.empty()) {
          empty_.wait();
        }
        T x = front();
        q_.pop();
        return x;
      }
      const T& front() const { return q_.front(); }
      bool empty() const { return q_.empty(); }
      void pop() { q_.pop(); }
      void clear() { while (!q_.empty()) q_.pop(); }
      const std::queue<T> &queue() const { return q_; }
    private:
      std::queue<T> q_;
      st_cond empty_;
  };

  template <typename T>
  class st_multichannel
  {
    public:
      void push(const T &x) {
        foreach (shared_ptr<st_channel<T> > q, qs) {
          q->push(x);
        }
      }
      st_channel<T> &subscribe() {
        shared_ptr<st_channel<T> > q(new st_channel<T>);
        qs.push_back(q);
        return *q;
      }
    private:
      vector<shared_ptr<st_channel<T> > > qs;
  };

  class st_intr_hub
  {
    public:
      virtual void insert(st_thread_t t) = 0;
      virtual void erase(st_thread_t t) = 0;
      virtual ~st_intr_hub() {};
  };

  class st_intr_cond : public st_intr_hub
  {
    public:
      virtual ~st_intr_cond() {}
      void insert(st_thread_t t) { threads.insert(t); }
      void erase(st_thread_t t) { threads.erase(t); }
      void signal() {
        foreach (st_thread_t t, threads) {
          st_thread_interrupt(t);
        }
        threads.clear();
      }
    private:
      std::set<st_thread_t> threads;
  };

  class st_intr_bool : public st_intr_hub
  {
    public:
      void insert(st_thread_t t) {
        if (b) st_thread_interrupt(t);
        else threads.insert(t);
      }
      void erase(st_thread_t t) { threads.erase(t); }
      void set() {
        b = true;
        foreach (st_thread_t t, threads) {
          st_thread_interrupt(t);
        }
        threads.clear();
      }
      void reset() {
        // If b is true, then any threads that join are immediately
        // interrupted, so the set must be empty.
        assert(!b || threads.empty());
        b = false;
      }
      operator bool() const { return b; }
    private:
      std::set<st_thread_t> threads;
      bool b;
  };

  class st_intr
  {
    public:
      st_intr(st_intr_hub &hub) : hub_(hub) { hub.insert(st_thread_self()); }
      ~st_intr() { hub_.erase(st_thread_self()); }
    private:
      st_intr_hub &hub_;
  };

  class st_group_join_exception : public std::exception
  {
    public:
      st_group_join_exception(const map<st_thread_t, std::exception> &th2ex) :
        th2ex_(th2ex) {}
      virtual ~st_group_join_exception() throw() {}
      virtual const char *what() const throw() {
        if (!th2ex_.empty() && s == "") {
          bool first = true;
          stringstream ss;
          typedef pair<st_thread_t, std::exception> p;
          foreach (p p, th2ex_) {
            ss << (first ? "" : ", ") << p.first << " -> " << p.second.what();
            first = false;
          }
          s = ss.str();
        }
        return s.c_str();
      }
    private:
      map<st_thread_t, std::exception> th2ex_;
      mutable string s;
  };

  class st_joining
  {
    NONCOPYABLE(st_joining)
    public:
      st_joining(st_thread_t t) : t_(t) {}
      ~st_joining() { st_join(t_); }
    private:
      st_thread_t t_;
  };

  class st_thread_group
  {
    public:
      ~st_thread_group() {
        map<st_thread_t, std::exception> th2ex;
        foreach (st_thread_t t, ts) {
          try { st_join(t); }
          catch (std::exception &ex) { th2ex[t] = ex; }
        }
        if (!th2ex.empty()) throw st_group_join_exception(th2ex);
      }
      void insert(st_thread_t t) { ts.insert(t); }
    private:
      std::set<st_thread_t> ts;
  };

  class eof_exception : public std::exception {
    const char *what() const throw() { return "EOF"; }
  };

  class st_reader
  {
    NONCOPYABLE(st_reader)
    public:
      st_reader(st_netfd_t fd, char *buf, size_t bufsize) :
        fd_(fd),
        buf_(buf, bufsize),
        start_(buf_.get()),
        end_(buf_.get())
      {}

      size_t unread() { return end_ - start_; }

      size_t rem() { return buf_.end() - end_; }

      sized_array<char> &buf() { return buf_; }

      void reset_range(char *start, char *end) {
        start_ = start;
        end_ = end;
      }

      void skip(size_t req, st_utime_t to = ST_UTIME_NO_TIMEOUT) {
        while (true) {
          if (unread() >= req) {
            // We have more unconsumed bytes than requested, so we're done.
            start_ += req;
            break;
          }

          // We have more requested bytes than unconsumed, so need to keep
          // reading.  Skip over bytes...
          req -= unread();
          // ...and reset pointers to discard current buffer.
          start_ = end_ = buf_.get();

          ssize_t res = checknnegerr(st_read(fd_, end_, rem(), to));
          end_ += res;

          // If we got a premature EOF.
          if (res == 0 && unread() < req) throw eof_exception();
        }
      }

      managed_array<char> read(size_t req, st_utime_t to = ST_UTIME_NO_TIMEOUT) {
        // Do we already have the requested data?
        if (unread() >= req) {
          managed_array<char> p(start_, false);
          start_ += req;
          return p;
        }

        // Handle large arrays specially.
        if (req > buf_.size()) {
          managed_array<char> p(new char[req], true);
          memcpy(p.get(), start_, unread());
          checkeqnneg(st_read_fully(fd_, p + unread(), req - unread(), to), static_cast<ssize_t>(req - unread()));
          start_ = end_ = buf_.get();
          return p;
        }

        // Shift things down if necessary.
        if (req > static_cast<size_t>(buf_.end() - end_)) {
          memmove(buf_.get(), start_, unread());
          size_t diff = start_ - buf_.get();
          start_ -= diff;
          end_ -= diff;
        }

        // Keep reading until we have enough.
        while (unread() < req) {
          ssize_t res = checknnegerr(st_read(fd_, end_, rem(), to));
          if (res == 0) break;
          else end_ += res;
        }

        // If we got a premature EOF.
        if (unread() < req)
          throw eof_exception();

        managed_array<char> p(start_, false);
        start_ += req;
        return p;
      }

      template<typename T>
      T read(st_utime_t to = ST_UTIME_NO_TIMEOUT)
      {
        size_t req = sizeof(T);

        // Do we already have the requested data?
        if (unread() >= req) {
          T x = *reinterpret_cast<const T*>(start_);
          start_ += req;
          return x;
        }

        assert(req <= buf_.size());

        // Shift things down if necessary.
        if (req > static_cast<size_t>(buf_.end() - end_)) {
          memmove(buf_.get(), start_, unread());
          size_t diff = start_ - buf_.get();
          start_ -= diff;
          end_ -= diff;
        }

        // Keep reading until we have enough.
        while (unread() < req) {
          ssize_t res = checknnegerr(st_read(fd_, end_, rem(), to));
          if (res == 0) break;
          else end_ += res;
        }

        // If we got a premature EOF.
        if (unread() < req)
          throw eof_exception();

        T x = *reinterpret_cast<const T*>(start_);
        start_ += req;
        return x;
      }

    private:
      st_netfd_t fd_;

      sized_array<char> buf_;

      char *start_;

      char *end_;
  };

}

#endif

---