10
\$\begingroup\$

The review I asked for yesterday was for a header only library.
Stream that opens an HTTP GET and then acts like a normal C++ istream

This has some limitations as it downloads all the data as part of the constructor. This can lead to delays during start up if the file is large (usually during REST communications the size is small so the header only version is sufficient).

But if you want larger files and don't want a delay here is a version that downloads the data in the background asynchronously. If the stream buffer underflows the next block (which has been accumulated in the background) is quickly swapped into the buffer and the main thread can continue processing.

The disadvantage is that this is no longer a header only library and you will need to link against the library.

https://github.com/Loki-Astari/ThorsStream

ThorsStream.h

#ifndef THORSANVIL_STREAM_THOR_STREAM_H
#define THORSANVIL_STREAM_THOR_STREAM_H

// Note "ThorsSimpleStream.h" is up fro review in a seprate question
// https://codereview.stackexchange.com/questions/38402/http-stream-a-stream-that-opens-an-http-get-and-then-acts-like-a-normal-c-istr
#include "ThorsSimpleStream.h"
#include <thread>
#include <iostream>

namespace ThorsAnvil
{
    namespace Stream
    {

class IThorStream: public std::istream
{
    // Class to handle the down-loading of all http connections in the background.
    class ThorStreamManager
    {
        public:
            // Everybody will use the same instance
            static ThorStreamManager& defaultManager()
            {
                static ThorStreamManager defaultManager;
                return defaultManager;
            }

            ThorStreamManager()
                : finished(false)
                , multi(curl_multi_init())
                , streamThread(&ThorStreamManager::eventLoop, this, multi != NULL)
            {
                if (multi == NULL)
                {   throw std::runtime_error("Failed to startup");
                }
            }
            ~ThorStreamManager()
            {
                // When destroying the object.
                // Make sure we first take care of any threads.
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    finished    = true;
                }
                cond.notify_one();
                streamThread.join();

                // Now we can clean up any resources.
                curl_multi_cleanup(multi);
            }

            // Interface to add/remove http requests.
            void addHTTPRequest(CURL* easy)
            {
                std::unique_lock<std::mutex> lock(mutex);
                curl_multi_add_handle(multi, easy);
            }
            void delHTTPRequest(CURL* easy)
            {
                std::unique_lock<std::mutex> lock(mutex);
                curl_multi_remove_handle(multi, easy);
            }

        private:
            // Event loop is run by the thread.
            // It handles all interactions with sockets.
            void eventLoop(bool ok);

            bool                        finished;
            CURLM*                      multi;
            std::mutex                  mutex;
            std::condition_variable     cond;
            std::thread                 streamThread;
    };
    // The stream buffer.
    class SocketStreamBuffer: public IThorSimpleStream::SimpleSocketStreamBuffer
    {
        public:
        typedef IThorSimpleStream::SimpleSocketStreamBuffer::traits_type    traits;
        typedef traits::int_type                                            int_type;

        SocketStreamBuffer(std::string const& url, std::function<void()> markStreamBad)
            : SimpleSocketStreamBuffer(url, false, false, markStreamBad)
        {
            /* Perform the request, res will get the return code */
            ThorStreamManager::defaultManager().addHTTPRequest(curl);
        }
        ~SocketStreamBuffer()
        {
            ThorStreamManager::defaultManager().delHTTPRequest(curl);
        }
        virtual int_type underflow()
        {
            {
                std::unique_lock<std::mutex> lock(mutex);
                empty = true;
            }
            curl_easy_pause(curl, CURLPAUSE_CONT);
            {
                std::unique_lock<std::mutex> lock(mutex);
                cond.wait(lock, [&empty, &open](){return !(empty && open);});
            }
            return empty ? EOF : buffer[0];
        }
        // Used by ThorStreamManager to mark that the stream has
        // finished downloading data.
        virtual void markAsDone()
        {
            std::unique_lock<std::mutex> lock(mutex);
            open    = false;
            cond.notify_one();
        }

    };

    SocketStreamBuffer    buffer;

    public:
        IThorStream(std::string const& url)
            : std::istream(NULL)
            , buffer(url, [this](){this->setstate(std::ios::badbit);})
        {
            std::istream::rdbuf(&buffer);
        }
};

    }
}

#endif

ThorsStream.cpp

#include "ThorsStream.h"

using namespace ThorsAnvil::Stream;


// Of OK is not true then the object
// will have thrown an exception so its
// its not safe to do anything but return
// immediately
void IThorStream::ThorStreamManager::eventLoop(bool ok)
{
    // If multi failed to initialize in the constructor
    // There is no point in continuing the thread.
    if (!ok)
    {   return;
    }

    // Can not let an exception escape a thread.
    // If it did it would cause the application
    // to terminate.
    try
    {
        std::unique_lock<std::mutex> lock(mutex);
        while(!finished)
        {
            // 1: Get the File descriptors that curl has ready.
            int     max_fd  = 0;
            fd_set  read;
            fd_set  writ;
            fd_set  exec;

            FD_ZERO(&read);
            FD_ZERO(&writ);
            FD_ZERO(&exec);

            CURLMcode   result  =  curl_multi_fdset(multi, &read, &writ, &exec, &max_fd);
            if (result != CURLM_OK) {throw std::runtime_error("curl_multi_fdset: Failed");}

            if (max_fd == 0)
            {
                // No file descriptors.
                // There is no point doing anything.
                // So wait until a new curl easy object is added.
                // Then re-start the loop.
                cond.wait(lock);
                continue;
            }

            // 2: Calculate the timeout we can use in select/poll
            timespec timeoutSpec = {0,0};
            if (max_fd == -1)
            {
                // curl special case.
                // We need to call curl_multi_perform() soon
                // so use a 100 millisecond timeout
                timeoutSpec.tv_sec  = 0;
                timeoutSpec.tv_nsec = 100 * 1000000;    // convert milli to nano
            }
            else
            {
                long timeout;
                result = curl_multi_timeout(multi, &timeout);
                if (result != CURLM_OK) {throw std::runtime_error("curl_multi_timeout: Failed");}
                if (timeout == -1)
                {   timeout = 300;
                }

                timeoutSpec.tv_sec  = timeout/1000;
                timeoutSpec.tv_nsec = timeout%1000 * 1000000; // convert milli to nano
            }

            // 3: Wait for activity on any of the file descriptors.
            int count;
            do
            {
                lock.unlock();
                count = pselect(max_fd+1, &read, &writ, &exec, &timeoutSpec, NULL);
                lock.lock();
            }
            while ((count == -1) && (errno == EINTR));
            if (count == -1) {throw std::runtime_error("pselect: Failed");}

            // 4: Handle all the CURL calllback routines.
            int running_handles;
            do
            {
                result = curl_multi_perform(multi, &running_handles);
            }
            while(result == CURLM_CALL_MULTI_PERFORM);
            if (result != CURLM_OK) {throw std::runtime_error("curl_multi_perform: Failed");}

            // 5: Handle all the messages generated by curl.
            //    Currently this is simply the fact that a stream has closed.
            CURLMsg*    message;
            int         msgs_in_queue;
            while((message = curl_multi_info_read(multi,  &msgs_in_queue)) != NULL)
            {
                if (message->msg == CURLMSG_DONE)
                {
                    CURL* easy = message->easy_handle;
                    char* priv;
                    if (curl_easy_getinfo(easy, CURLINFO_PRIVATE, &priv) == CURLE_OK && priv != NULL)
                    {
                        IThorStream::SocketStreamBuffer* owner = reinterpret_cast<IThorStream::SocketStreamBuffer*>(priv);
                        owner->markAsDone();
                    }
                }
            }
        }
    }
    catch(std::exception const& e)
    {
        // Log error:
        std::cerr << "Exception in Thread:\n\t" << e.what() << "\n";
    }
    catch(...)
    {
        // Log error:
        std::cerr << "Exception in Thread:\n\tUNKNOW (...)\n";
    }
    // Ignore error
}

Unfortunately I used the select() version of the CURL interface. I have plans to re-write this to use the poll() version of the interface. But that's another day.

A simple example using the above stream and my previous JSON parser library to connect to a REST interface and download data directly into a C++ class object without parsing the JSON directly.

https://gist.github.com/Loki-Astari/8201956

\$\endgroup\$
3
\$\begingroup\$
#include "ThorsSimpleStream.h"

There's no ThorsSimpleStream identifier used in the header: instead, "ThorsSimpleStream.h" declares class IThorSimpleStream.

I prefer it if the filename of the header matches the name of identifier which it declares.


IThorSimpleStream declares friend class IThorStream.

I prefer it if friend classes are defined in the same header file (this suggestion comes from Lakos, summarized as "no long-distance friendship").

SimpleSocketStreamBuffer seems to be a private member of IThorStream and IThorSimpleStream. Instead of declaring friendship between IThorStream and IThorSimpleStream, perhaps you could simply define SimpleSocketStreamBuffer in its own header file.


class IThorStream

I'm used to I being used for designate 'interfaces': or pure abstract classes in C++. I don't see why you're prefixing your class names with I.


You could move more out of the header and into the CPP file: the definition/implementation of SocketStreamBuffer methods, and the whole ThorStreamManager class declaration.

Users of your header file would then have less to read, and fewer headers such as <thread> included into their code.

You could hide the whole SocketStreamBuffer declaration too, using the 'cheshire cat' aka 'pimpl' idiom (i.e. by making it an opaque pointer).


           lock.unlock();
           count = pselect(max_fd+1, &read, &writ, &exec, &timeoutSpec, NULL);
           lock.lock();

Might you want to do if (finished) break; after re-acquiring the lock, in case the system is shutting down / waiting to shutdown?


catch(std::exception const& e)

This is outside the while loop so the thread will die so the ThorStreamManager is dead. You might like to move that into the while loop, or re-create the ThorStreamManager in case there's a next request.


(I haven't learned the CURL API so can't review your eventLoop in any detail.)

\$\endgroup\$
  • \$\begingroup\$ IThorStream is following the conventions of the standard library. istream, ifstream and ostream, ofstream. I don't use the I is for interface convention as I feel it is redundant. \$\endgroup\$ – Martin York Mar 24 '14 at 0:50
  • \$\begingroup\$ I used the name ThorsSimpleStream.h as one day I may add the class OThorSimpleStream. \$\endgroup\$ – Martin York Mar 24 '14 at 1:00

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.