1
\$\begingroup\$

I'm a bit of a neophyte when it comes to C++, and so I'd like some feedback regarding a recent project.

The code sits on a Raspberry Pi and streams camera data over TCP on a specified port.

The MJPGWriter class:

#include <unistd.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#define PORT        unsigned short
#define SOCKET    int
#define HOSTENT  struct hostent
#define SOCKADDR    struct sockaddr
#define SOCKADDR_IN  struct sockaddr_in
#define ADDRPOINTER  unsigned int*
#define INVALID_SOCKET -1
#define SOCKET_ERROR   -1
#define TIMEOUT_M       200000
#define NUM_CONNECTIONS 10

#include <pthread.h>
#include <iostream>
#include <stdio.h>
#include "opencv2/opencv.hpp"

using namespace cv;
using namespace std;

struct clientFrame {
    uchar* outbuf;
    int outlen;
    int client;
};

struct clientPayload {
    void* context;
    clientFrame cf;
};

class MJPGWriter{
    SOCKET sock;
    fd_set master;
    int timeout;
    int quality; // jpeg compression [1..100]
    std::vector<int> clients;

    pthread_mutex_t mutex_client = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_t mutex_cout = PTHREAD_MUTEX_INITIALIZER;

    int _write(int sock, char *s, int len)
    {
        if (len < 1) { len = strlen(s); }
        {
            int retval = ::send(sock, s, len, 0);
            return retval;
        }
    }

public:

    MJPGWriter(int port = 0)
        : sock(INVALID_SOCKET)
        , timeout(TIMEOUT_M)
        , quality(30)
    {
        FD_ZERO(&master);
        if (port)
            open(port);
    }

    ~MJPGWriter()
    {
        release();
    }

    bool release()
    {
        if (sock != INVALID_SOCKET)
            shutdown(sock, 2);
        sock = (INVALID_SOCKET);
        return false;
    }

    bool open(int port)
    {
        sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

        SOCKADDR_IN address;
        address.sin_addr.s_addr = INADDR_ANY;
        address.sin_family = AF_INET;
        address.sin_port = htons(port);
        if (bind(sock, (SOCKADDR*)&address, sizeof(SOCKADDR_IN)) == SOCKET_ERROR)
        {
            cerr << "error : couldn't bind sock " << sock << " to port " << port << "!" << endl;
            return release();
        }
        if (listen(sock, NUM_CONNECTIONS) == SOCKET_ERROR)
        {
            cerr << "error : couldn't listen on sock " << sock << " on port " << port << " !" << endl;
            return release();
        }
        FD_SET(sock, &master);
        return true;
    }

    bool isOpened()
    {
        return sock != INVALID_SOCKET;
    }

    static void* listen_Helper(void* context)
    {
        ((MJPGWriter *)context)->Listener();
    }

    static void* writer_Helper(void* context)
    {
        ((MJPGWriter *)context)->Writer();
    }

    static void* clientWrite_Helper(void* payload)
    {
        void* ctx = ((clientPayload *)payload)->context;
        struct clientFrame cf = ((clientPayload *)payload)->cf;
        ((MJPGWriter *)ctx)->ClientWrite(cf);
    }

private:
    void Listener();
    void Writer();
    void ClientWrite(clientFrame &cf);
};

The Listener function loops indefinitely while the program runs, listening in to new clients as they are connected. On connection, it sends the header.

void
MJPGWriter::Listener()
{
    fd_set rread;
    SOCKET maxfd;
    while (true)
    {
        rread = master;
        struct timeval to = { 0, timeout };
        maxfd = sock + 1;

        int sel = select(maxfd, &rread, NULL, NULL, &to);
        if (sel > 0) {
            for (int s = 0; s < maxfd; s++)
            {
                if (FD_ISSET(s, &rread) && s == sock)
                {
                    int         addrlen = sizeof(SOCKADDR);
                    SOCKADDR_IN address = { 0 };
                    SOCKET      client = accept(sock, (SOCKADDR*)&address, (socklen_t*)&addrlen);
                    if (client == SOCKET_ERROR)
                    {
                        cerr << "error : couldn't accept connection on sock " << sock << " !" << endl;
                        return;
                    }
                    maxfd = (maxfd>client ? maxfd : client);
                    pthread_mutex_lock(&mutex_cout);
                    cout << "new client " << client << endl;
                    pthread_mutex_unlock(&mutex_cout);
                    pthread_mutex_lock(&mutex_client);
                    _write(client, (char*)"HTTP/1.0 200 OK\r\n", 0);
                    _write(client, (char*)
                        "Server: Mozarella/2.2\r\n"
                        "Accept-Range: bytes\r\n"
                        "Connection: close\r\n"
                        "Max-Age: 0\r\n"
                        "Expires: 0\r\n"
                        "Cache-Control: no-cache, private\r\n"
                        "Pragma: no-cache\r\n"
                        "Content-Type: multipart/x-mixed-replace; boundary=mjpegstream\r\n"
                        "\r\n", 0);
                    clients.push_back(client);
                    pthread_mutex_unlock(&mutex_client);
                }
            }
        }
        usleep(10);
    }
}

The Writer function also continues to loop, getting the information from the camera, and spawns a thread for each client to write the information to.

void
MJPGWriter::Writer()
{
    VideoCapture cap;
    bool ok = cap.open(0);
    if (!ok)
    {
        printf("no cam found ;(.\n");
        pthread_exit(NULL);
    }

    while (cap.isOpened() && this->isOpened())
    {
        Mat frame;
        cap >> frame;
        pthread_t threads[NUM_CONNECTIONS];
        int count = 0;

        std::vector<uchar>outbuf;
        std::vector<int> params;
        params.push_back(CV_IMWRITE_JPEG_QUALITY);
        params.push_back(quality);
        imencode(".jpg", frame, outbuf, params);
        int outlen = outbuf.size();

        pthread_mutex_lock(&mutex_client);
        std::vector<int>::iterator begin = clients.begin();
        std::vector<int>::iterator end = clients.end();
        pthread_mutex_unlock(&mutex_client);

        for (std::vector<int>::iterator it = begin; it != end; ++it, ++count)
        {
            if (count > NUM_CONNECTIONS)
                break;
            struct clientPayload cp = { (MJPGWriter*)this, { &outbuf[0], outlen, *it } };
            pthread_create(&threads[count-1], NULL, &MJPGWriter::clientWrite_Helper, &cp);
        }
        for (; count > 0; count--)
        {
            pthread_join(threads[count-1], NULL);
        }
        usleep(10);
    }
}

The Client Write function then writes this information to the connected client.

void
MJPGWriter::ClientWrite(clientFrame & cf)
{
    char head[400];
    sprintf(head, "--mjpegstream\r\nContent-Type: image/jpeg\r\nContent-Length: %lu\r\n\r\n", cf.outlen);
    _write(cf.client, head, 0);
    int n = _write(cf.client, (char*)(cf.outbuf), cf.outlen);
    if (n < cf.outlen)
    {
        pthread_mutex_lock(&mutex_client);
        cerr << "kill client " << cf.client << endl;
        clients.erase(std::remove(clients.begin(), clients.end(), cf.client));
        ::shutdown(cf.client, 2);
        pthread_mutex_unlock(&mutex_client);
    }
    pthread_exit(NULL);
}

Main: intiailise with port 7777 and spawn the two controllers:

int
main()
{
    MJPGWriter test(7777);

    pthread_t thread_listen, thread_write;
    pthread_create(&thread_listen, NULL, &MJPGWriter::listen_Helper, &test);
    pthread_create(&thread_write, NULL, &MJPGWriter::writer_Helper, &test);

    pthread_join(thread_listen, NULL);
    pthread_join(thread_write, NULL);

    exit(0);

}

The programme works to some extent. I can connect to the IP address and port from another computer in the network, with either Firefox or VLC, and get a video stream from the camera.

However, it isn't very stable, and multiple connections seem to slow it down, sometimes appearing to serve only one client at a time. Also, quite often when a client disconnects, the code ends without error, instead of continuing to serve the remaining clients/wait for new clients.

\$\endgroup\$
3
\$\begingroup\$

One of your problems is that you are passing by parameter to the ClientWrite's threads a reference to a struct that you're replacing in the 'for loop', so all the threads will have the same struct (cf.client, cf.outbuf and cf.outlen) so you're sending all the data to the last cf.client.

You should replace this code:

struct clientPayload cp = { (MJPGWriter*)this, { &outbuf[0], outlen, *it } };
pthread_create(&threads[count-1], NULL, &MJPGWriter::clientWrite_Helper, &cp);

With something like this:

struct clientPayload *cp = new clientPayload({ (MJPGWriter*)this, { outbuf.data(), outlen, *it } });
pthread_create(threads[count], NULL, &MJPGWriter::clientWrite_Helper, cp);

You should also save all the structs in a vector, for example, so you will be able to delete all the structs when they have been used.

I have created a repo in my github where I'm trying to improve your server because I want to stream the video of one of my OpenCV programs. Thanks you for sharing your code. It was so helpful to me.

I hope this answer to be helpful for you too.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.