Dynamic boost threads running methods of objects stored in a vectors with shared pointers

Question

I have a camera class which is an abstraction of a network camera. The class has a method rxImageThread() which is an image server, essentially. I want to run n servers in n threads for n cameras where n is dynamically set (from a config file).

The camera objects are stored in a vector:

std::vector<boost::shared_ptr<camera> > cameras;

I create the camera objects in a loop:

cameras.push_back( boost::shared_ptr<camera> (new camera(ip, controlPort, dataPort, imagePort, name)) );

I have a vector to store the threads:

std::vector<boost::shared_ptr<boost::thread> > threads;

In a loop, I want to enable the rxImageThread()'s of each camera to run in a thread:

for(int i = 0;i<cameras.size();i++){
   threads.push_back( boost::shared_ptr<boost::thread> (new boost::thread(&camera::rxImageThread, &cameras[i]) );
   cameras[i]->startRx();
}

However, I get an error along the lines of:

‘void (camera::)()’ incompatible with object type ‘boost::shared_ptr<camera>’

Which suggests that I should use code similar to:

threads.push_back( boost::shared_ptr<boost::thread> (new boost::thread(std::mem_fn(&camera::rxImageThread),&cameras[i]  ));

However, this gets the same errors.

I'm a little lost as to what's going on here, let alone how to fix it.

Answer 1

a shared_ptr<> is not an address. you have to call .get() on it to get the object address.

such as:

threads.push_back( boost::shared_ptr<boost::thread> (new boost::thread(&camera::rxImageThread, cameras[i].get()) );

Looking further than solving this problem, I am wondering whether it might be cleaner to have a controller class that owns the camera and controls the thread lifetime. This way you can guarantee that the camera is not deleted before the owning thread terminates.

EDIT:

here's a complete program that demonstrates the idea. Thread safe starting and stopping of cameras. no memory leaks or possibility of threads outliving cameras:

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

#include <vector>
#include <iostream>
#include <algorithm>

using namespace std;


namespace {
    boost::mutex _cout_mutex;
}
// a cut-down camera class

struct camera
{
    camera(std::string hostname, unsigned short port)
    : _hostname(hostname)
    , _port(port)
    {}

    std::string description() const {
        std::ostringstream ss;
        ss << _hostname << ':' << _port;
        return ss.str();
    }

    std::string _hostname;
    unsigned short _port;
};



struct camera_controller
{
    camera_controller(const std::string& hostname, unsigned short port)
    : _camera(hostname, port)
    , _stop(false)
    {}

    void start() {
        boost::unique_lock<boost::mutex> outlock(_cout_mutex);
        cout << "starting " << _camera.description() << endl;
        outlock.unlock();

        assert(!_thread_ptr);
        _stop = false;
        _thread_ptr.reset(new boost::thread(boost::bind(&camera_controller::camera_thread, this)));
    }

    void stop() {
        if (_thread_ptr) {
            boost::unique_lock<boost::mutex> outlock(_cout_mutex);
            cout << "stopping " << _camera.description() << endl;
            outlock.unlock();

            notify_stop();
            _thread_ptr->join();
            _thread_ptr.reset();
        }
    }

    ~camera_controller()
    {
        stop();
    }

private:

    void camera_thread()
    {
        while (!time_to_stop())
        {
            // do things with the camera
            boost::unique_lock<boost::mutex> outlock(_cout_mutex);
            cout << "working: " << _camera.description() << endl;
            outlock.unlock();
            boost::this_thread::sleep_for(boost::chrono::milliseconds(1000));

        }
        boost::unique_lock<boost::mutex> outlock(_cout_mutex);
        cout << "shutting down: " << _camera.description() << endl;
        outlock.unlock();
    }

    void notify_stop() {
        boost::unique_lock<boost::mutex> outlock(_cout_mutex);
        cout << "notify stop for " << _camera.description() << endl;
        outlock.unlock();
        _stop = true;
        // would normally signal a condition variable here
    }

    bool time_to_stop() const {
        // test the stop condition
        return _stop;
    }

private:
    camera _camera;
    boost::shared_ptr<boost::thread> _thread_ptr;

    // this could be a condition variable in a real application
    bool _stop;
};


struct application
{
    typedef boost::shared_ptr<camera_controller> control_ptr;

    control_ptr add_camera(const std::string& host, unsigned short port)
    {
        control_ptr p(new camera_controller(host, port));
        p->start();

        // tightest lock possible on shared resources
        boost::unique_lock<boost::mutex> lock(_cameras_mutex);
        _camera_controllers.push_back(p);
        return p;
    }

    // for example...
    void destroy_camera_by_index(size_t i)
    {
        boost::unique_lock<boost::mutex> lock(_cameras_mutex);
        control_ptr cam_ptr = _camera_controllers[i];
        _camera_controllers.erase(_camera_controllers.begin() + i);
        lock.unlock();

        // note - this blocks until the camera thread is stopped
        cam_ptr->stop();
    }

    size_t camera_count() const {
        boost::unique_lock<boost::mutex> lock(_cameras_mutex);
        return _camera_controllers.size();
    }


private:

    std::vector<control_ptr> _camera_controllers;
    mutable boost::mutex _cameras_mutex;
};

int main()
{
    application app;
    app.add_camera("camera_a", 0);
    app.add_camera("camera_b", 1);
    app.add_camera("camera_c", 2);
    app.add_camera("camera_d", 3);
    app.add_camera("camera_e", 4);

    while(app.camera_count() > 2) {
        boost::this_thread::sleep_for(boost::chrono::seconds(2));
        size_t i = rand() % app.camera_count();
        app.destroy_camera_by_index(i);
    }

    boost::this_thread::sleep_for(boost::chrono::seconds(2));

}