[Boost-users] boost::threads Am I on the right track?

From: The Grumpiest Man You Know (mrgrumpyguts_at_[hidden])
Date: 2005-03-05 21:15:49

I've had a minor epiphany, or at least I think so, in use of
boost::threads.  I would be very grateful if someone who actually knew
something about C++ and boost's threads support would glance at the
attached code.  Any comment would be appreciated especially if there's
sign that I'm horribly misusing the scraps I've picked up.  I have to
say that once I stopped trying to relate the pthreads C API to the
bits of the docs I could understand my reaction was, "Sweet!"  I think
the way the threading and locking "disappear" is ... really elegant.

Thanks very much for your time and again if I'm OT please tell me
where to go, even if it is to hell. :)

------------

#include <iostream>
#include <vector>
#include <queue>

#include <boost/thread/thread.hpp>
#include <boost/function.hpp>

using namespace std;
using namespace boost;

mutex::mutex io;

struct Throttle
{
  Throttle(void (*fn)(), int svcQLen = 5) :
    population(1), load(0), acceptable(svcQLen), service(fn) {
  };
  ~Throttle() {
    cout << "load is " << load << endl;
    cout << "Population " << population << endl;
  }

  void itsBusy() {
    mutex::scoped_lock myPermission(use);
    if (++load > acceptable) {
      workers.create_thread(service);
      population++;
      load = 0;
    };
  };

  void observe() {
    workers.join_all();
  };

private:
  mutex use;
  int population;
  int load;
  int acceptable;
  void (*service)();
  thread_group workers;
};

template <class T>
struct RequestQ : public queue<T> {
  typedef T value_type;
  typedef queue<T> cBase;

  bool empty() {
    mutex::scoped_lock permit(use);

    return cBase::empty();
  }
  value_type pop() {
    mutex::scoped_lock permit(use);

    value_type it = cBase::front();
    cBase::pop();

    mutex::scoped_lock doing(io);
    cout << "Q -> " << it << endl;

    return it;
  }
  void extend(unsigned int by) {
    mutex::scoped_lock permit(use);
    value_type next;

    if (cBase::empty()) {
      next = 0;			// Yeah, assumes integral type conversion
    } else {
      next = cBase::back();
    }

    while(by--) {
      push(*new value_type(++next));
    }
  }

private:
  void push(const value_type& it) {
    mutex::scoped_lock doing(io);
    cout << "Q <- " << it << endl;

    cBase::push(it);
  }

  mutex use;
};

/*
 * We have to initialise this before creating any threads but that's
 * OK 'cos this IS the way we'll create threads.
 */
Throttle *manager;

RequestQ<int> jobs;

void
go()
{
  for(int range = 0; !jobs.empty(); range = 0) {
    int mine = jobs.pop();

    if (!jobs.empty()) {
      manager->itsBusy();
    }

    {
      mutex::scoped_lock doing(io);
      bool term = false;

      if (!(mine % 3)) {
        cout << "fizz ";
        range=1;
        term = true;
      }
      if (!(mine % 5)) {
        cout << "buzz ";
        range += 1;
        term = true;
      }
      if (!(mine % 7)) {
        cout << "pop! ";
        range = 0;
        term = true;
      }
      if (!(mine % 11)) {
        cout << "puff";
        range = 2 * (range + 1);
        term = true;
      }

      if (term) {
        cout <<endl;
      }
    }

    if (range) {
      jobs.extend(range);
      sleep(range);
    }
  }
}

int
main(int argc, char **argv)
{
  manager = new Throttle(&go);

  jobs.extend(argc * 10);

  go();

  manager->observe();
  delete manager;
}

-- 
Blue Skies