8
\$\begingroup\$

Need a way to test a web crawler.

Threw together this web server that will replay previously saved pages from a real web server.

Headers

#include <stdexcept>
#include <memory>
#include <map>
#include <iostream>
#include <sstream>
#include <fstream>
#include <string>
#include <algorithm>

#include <cstdio>
#include <cstdlib>
#include <cerrno>
#include <strings.h>

#include <pthread.h>
#include <netdb.h>
#include <netinet/in.h>



extern "C" void* workerThread(void*);

Multi Threading Stuff

class SimpleThread
{
    pthread_t       thread;

    SimpleThread(SimpleThread const&);
    SimpleThread& operator=(SimpleThread const&);

    public:
        SimpleThread()
        {
            if (::pthread_create(&thread, NULL, workerThread, NULL) != 0)
            {
                throw std::runtime_error("Failed to Start Thread");
            }
        }
        ~SimpleThread()
        {
            void*   result;
            if (::pthread_join(thread, &result) != 0)
            {
                throw std::runtime_error("Failed to Join Thread");
            }
        }
};

class SimpleCondition;
class SimpleMutex
{
    SimpleMutex(SimpleMutex const&);
    SimpleMutex operator=(SimpleMutex const&);

    friend class SimpleCondition;
    pthread_mutex_t     mutex;

    public:
        SimpleMutex()
        {
            if (::pthread_mutex_init(&mutex, NULL) != 0)
            {
                throw std::runtime_error("Failed to Create Mutex");
            }
        }
        ~SimpleMutex()
        {
            ::pthread_mutex_destroy(&mutex);
        }
        void lock()
        {
            if (::pthread_mutex_lock(&mutex) != 0)
            {
                throw std::runtime_error("Filed to lock mutex");
            }
        }
        void unlock()
        {
            if (::pthread_mutex_unlock(&mutex) != 0)
            {
                throw std::runtime_error("Faile to unlock mutex");
            }
        }
};
class Locker
{
    SimpleMutex&    m;
    public:
        Locker(SimpleMutex& m)
            : m(m)
        {
            m.lock();
        }
        ~Locker()
        {
            m.unlock();
        }
};

class SimpleCondition
{
    SimpleCondition(SimpleCondition const&);
    SimpleCondition operator=(SimpleCondition const&);

    pthread_cond_t cond;
    public:
        SimpleCondition()
        {
            if (::pthread_cond_init(&cond, NULL) != 0)
            {
                throw std::runtime_error("Failed to init condition variable");
            }
        }
        ~SimpleCondition()
        {
            ::pthread_cond_destroy(&cond);
        }
        void wait(SimpleMutex& mutex)
        {
            if (::pthread_cond_wait(&cond, &mutex.mutex) != 0)
            {
                throw std::runtime_error("Failed to wait on mutex");
            }
        }
        void notify()
        {
            if (::pthread_cond_signal(&cond) != 0)
            {
                throw std::runtime_error("Failed to notify");
            }
        }
        void releaseAll()
        {
            if (::pthread_cond_broadcast(&cond) != 0)
            {
                throw std::runtime_error("Failed to release all");
            }
        }
};

Very Simple Job and Queue

/* Job object
 *  sockfd:     The stream file descritor to send the reply on.
 *  fileName:   The name of the file with data to be sent.
 */
struct Job
{
    int         sockfd;
    std::string fileName;
    Job*        next;

    Job(int sockfd, std::string const& fileName)
        : sockfd(sockfd)
        , fileName(fileName)
        , next(NULL)
    {}
    ~Job()
    {
        if (sockfd)
        {
            ::close(sockfd);
        }
    }
};

class WorkQueue
{
    Job*            head;
    Job*            tail;
    SimpleMutex     access;
    SimpleCondition condition;
    bool            finished;

    public:
        WorkQueue()
            : head(NULL)
            , tail(NULL)
            , finished(false)
        {}
        ~WorkQueue()
        {
            finished = true;
            condition.releaseAll();
            while(head)
            {
                Job*    item    = head;
                head = item->next;
                delete item;
            }
        }
        void addItem(Job* another)
        {
            {
                Locker      lock(access);
                if (head == NULL)
                {
                    head        = another;
                }
                else
                {
                    tail->next  = another;
                }
                tail        = another;
            }
            condition.notify();
        }
        Job* getItem()
        {
            Locker      lock(access);
            while(head == NULL)
            {
                condition.wait(access);
                if (finished)
                {
                    return NULL;
                }
            }
            Job*        result = head;
            head = result->next;
            if (result == tail)
            {
                tail = NULL;
            }
            return result;
        }
};

The Socket

class SimpleSocket
{
    int sockfd;
    public:
        SimpleSocket(int port)
        {
            sockfd = ::socket(AF_INET, SOCK_STREAM, 0);
            if (sockfd < 0)
            {
                throw std::runtime_error("Failed to create socket");
            }

            struct sockaddr_in serv_addr;
            ::bzero((char *) &serv_addr, sizeof(serv_addr));

            serv_addr.sin_family        = AF_INET;
            serv_addr.sin_addr.s_addr   = INADDR_ANY;
            serv_addr.sin_port          = htons(port);

            if (::bind(sockfd, reinterpret_cast<sockaddr*>(&serv_addr), sizeof(serv_addr)) < 0)
            {
                throw std::runtime_error("Failed to bind to port");
            }
        }
        ~SimpleSocket()
        {
            ::close(sockfd);
        }
        int waitForConnection()
        {
            int result = ::listen(sockfd,5);
            if (result < 0)
            {
                throw std::runtime_error("Listen failed");
            }
            sockaddr_in     cli_addr;
            socklen_t       clilen      = sizeof(cli_addr);
            int             newsockfd   = ::accept(sockfd, reinterpret_cast<sockaddr*>(&cli_addr), &clilen);
            if (newsockfd < 0)
            {
                throw std::runtime_error("accept failed");
            }
            return newsockfd;
        }
};

Class for abstracting Pages

/* The set of pages the can be returned */
class Pages
{
    public:
        // Note 1: page contains the server name and page.
        // See:    getPageFromRequest()
        // Note 2: A response is preset in the files <filaName>.head <filaName>.body
        // See:    workerThread()
        Job* getJob(int sockfd, std::string page) const
        {
            std::string  fileName = std::string("/mnt/Pages/") + page;

            struct stat buffer;
            bool ok = ::stat((fileName + ".head").c_str(), &buffer) != 0)
                  &&  ::stat((fileName + ".body").c_str(), &buffer) != 0)
            if (!ok)
            {
                // Note: Invalid will not clash with saved pages.
                //       because pages are in sub-directories based on
                //       the server name.
                return new Job(sockfd, "/mnt/Pages/Invalid");
            }
            return new Job(sockfd, fileName);
        }
};

Helper function for reading/writing to stream

/* Copy a buffer to stream */
void sendToClient(int sockfd, char* buffer, int size)
{
    std::size_t write = 0;
    while(write != size)
    {
        std::size_t out = ::write(sockfd, buffer + write, size - write);
        if (out == -1 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR))
        {
            continue;
        }
        if (out == -1)
        {
            throw std::runtime_error("Failed to write to socket");
        }
        write += out;
    }
}

/* Copy a File to a stream */
void writeFile(int sockfd, std::string const& fileName)
{
    char                buffer[4096];
    std::ifstream       file(fileName.c_str());
    std::size_t         count;

    do
    {
        file.read(buffer, 4096);
        count = file.gcount();
        sendToClient(sockfd, buffer, count);
    }
    while(count > 0);
}
/* Read the request from the socket */
std::string getPageFromRequest(int connection)
{
    char buffer[4097];
    std::size_t read = 0;
    while(true)
    {
        std::size_t atual = ::read(connection, buffer + read, 4096 - read);
        if (atual == 0)
        {
            break;
        }
        if (atual == -1 && (errno == EAGAIN || errno == EINTR))
        {
            continue;
        }
        if (atual == -1)
        {
            throw std::runtime_error("Read Error");
        }
        read += atual;
        if (read == 4096)
        {
            break;
        }
        if ((read > 3) && buffer[read - 3] == '\n' && buffer[read - 2] == '\r' && buffer[read - 1] == '\n')
        {
            break;
        }
    }
    static std::string const marker = "Host: ";

    char* pageSt  = buffer + 4;
    char* pageEnd = std::find(pageSt, buffer + read, ' ');
    char* servSt  = std::search(buffer, buffer + read, marker.begin(), marker.end());
    char* servEnd = std::find(servSt + marker.size(), buffer + read, '\r');

    std::string req = std::string(servSt + marker.size(), servEnd) + std::string(pageSt, pageEnd);
    return std::string(pageSt, pageEnd);
}

The work part of the server

WorkQueue       work;
void* workerThread(void*)
{
    Job* job;
    try
    {
        while(job = work.getItem())
        {
            std::auto_ptr<Job>    workItem(job);
            if (job->sockfd)
            {
                writeFile(workItem->sockfd, workItem->fileName + ".head");
                writeFile(workItem->sockfd, workItem->fileName + ".body");
            }
        }
    }
    catch(...){}

    return NULL;
}

int main()
{
    SimpleThread    workers[8];
    SimpleSocket    socket(80);
    Pages           pages;

    int connection;
    while(connection = socket.waitForConnection())
    {
        std::string page = getPageFromRequest(connection);
        work.addItem(pages.getJob(connection, page));
    }

}
\$\endgroup\$
4
\$\begingroup\$

First of all, I really liked your C++ code and even though I am not a professional I think there are some points to consider.

To the future!

It is 2016 so I would say we can safely use C++11 today (at least if you are not held back by embedded, an ancient compiler, etc.). There being at least three major compilers with C++11 support this shoud not be a problem.

I went through your code from top to bottom and the first thing was SimpleThread. It is good but we do not need it because there is std::thread

int main() {
  std::vector<std::thread> workers{std::thread::hardware_concurrency()};
  ...
  for(auto& worker : workers) {
    worker = std::thread(workerThread, nullptr);
  }
}

So instead of using 8 as a safe assumption for our threads I used hardware_concurrency() to ask the system how many cores it has/we can utilize. The constructor of threadis more generic than SimpleThread and takes the function as a parameter. thread being templated we can safely use a C++ function with arbitrary arguments which allows us to change workerThread to

void workerThread(WorkQueue& work)

and enables us to remove the global variable and the unneccessary return null which would be in C++11 terms return nullptr. Because we pass a reference we have to use thread(workerThread, std::ref(work)) to allow the template magic to find the right workerThread.

Going forward, your classes SimpleCondition and SimpleMutex are redundant as well because there are std::condition_variable and std::mutex which do pretty much the same thing.

Instead of releaseAll() and notify we use notify_all() and notify_one() (which describes their use cases and similarity better). To lock the mutex we use std::unique_lock<std::mutex> lock{access} which works like your Lock. A real difference is in condition.wait() because it takes the lock and not the mutex as a parameter condition.wait(lock).

I know everyone (in C/C++) likes to implement their linked lists, but why bother when there is no real benefit. It takes more work, it increases cache misses because the objects are scattered in the memory and in your implementation it locks Job to this implementation without a reason. Job should just be a data structure regardless of the container it is put in. So I removed Job* next from it.

To accomodate WorkQueue has to change as well. I decided to put a std::queue in it because it is exactly what we need.

class WorkQueue {
  std::queue<std::unique_ptr<Job>> jobs;
  std::mutex access;
  std::condition_variable condition;
  bool finished;

  public:
    WorkQueue()
      : finished(false)
    {}
    ~WorkQueue() {
      finished = true;
      condition.notify_all();
    }
    void addItem(std::unique_ptr<Job> another) {
      {
        std::unique_lock<std::mutex> lock{access};
        jobs.push(std::move(another));
      }
      condition.notify_one();
    }
    std::unique_ptr<Job> getItem() {
      std::unique_lock<std::mutex> lock{access};
      condition.wait(lock, [this] () { return !jobs.empty() || finished; });
      if (finished) {
        return std::unique_ptr<Job>(nullptr);
      }
      std::unique_ptr<Job> nextJob = std::move(jobs.front());
      jobs.pop();
      return nextJob;
    }
};

While I was at it I changed Job* to a std::unique_ptr. I saw you used std::auto_ptr, do not use it, it is deprecated and will be removed in C++17. unique_ptr is not copy-constructible/-assignable therefore we have to move it into jobs in addItem and out of it in getItem.

A more drastic change is the wait logic. I could have used condition.wait(lock) in a while loop but there is an overload of wait() which takes a predicate to check if waiting is over and it fits this perfectly.

In your getJob I created the unique_ptr using std::make_uniqe() which is the highly recommended way to create unique_ptrs. While I was at it I made it static because it removes the need to instantiate Pages.

Destroy all the things!

Wait what?! Even though I wanted to show you all these neat ways which simplify your code in the end I deleted all your thread code and most of my changes because there is a simpler solution to your problem.

You know that IO takes time and blocks so you thought of a way to make it asynchronous. The common approach, as you did, is to implement some kind of threadpool logic with workers and creators. But there is an alternative, std::async. It encapsulates the asynchronous logic and is required to act like if it was a std::thread. Most implementations problably use a threadpool so it is exactly what you did, just less work for you.

I removed SimpleQueue and changed workerThread (not being a thread anymore)

void handleJob(std::unique_ptr<Job> job) {
  if (job->sockfd) {
    writeFile(job->sockfd, job->fileName + ".head");
    writeFile(job->sockfd, job->fileName + ".body");
  }
}

int main() {
  std::vector<std::future<void>> pending_futures;
  SimpleSocket socket(80);

  int connection;
  while(connection = socket.waitForConnection()) {
    std::string page = getPageFromRequest(connection);
    pending_futures.push_back(std::async(std::launch::async, handleJob, Pages::getJob(connection, page)));
  }
}

The call to async returns a std::future which gives a way to retrieve the return value of our operation. Moreover, it enables you to join the operation, if you want that or not. Because the destructor automatically joins the operation if it is not finished we have to keep the futures around. One might want to clean these up, but this is left as an exercise for the reader ;)

Going experimental

Huh C++11/C++14 is not enough? How about the Filesystem TS which will someday (might be C++17) be in the standard. Nevertheless there are implementations in gcc (5.x+ I guess) and VS (2015) (clang says no work started) which you can use with #include <experimental/filesystem>.

It has a clean and portable API to work with filesystems, even though I only used it in one place I think it is more descriptive.

namespace fs = std::experimental::filesystem;
...
bool ok = fs::is_regular_file(fileName + ".head") && fs::is_regular_file(fileName + ".body");

I am not sure if this is what you meant with your stat call because you only check for errors. But a nonexistent file does not return an error code. But from the context I think a check with is_regular_file() is a reasonable check to make.

Theoretically there is also a Networking TS which should be tested by example implementation in the compilers like the Filesystem TS but I could not find any compiler implementing it. With this TS your network code could be simplified as well, but this is for another day.

The final code

#include <stdexcept>
#include <memory>
#include <map>
#include <iostream>
#include <sstream>
#include <fstream>
#include <string>
#include <algorithm>
#include <thread>
#include <vector>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <future>
#include <experimental/filesystem>

#include <cstring> // strerror
#include <cstdio>
#include <cstdlib>
#include <cerrno>
#include <strings.h>

#include <unistd.h> // ::close

#include <netdb.h>
#include <netinet/in.h>

using namespace std::literals;
namespace fs = std::experimental::filesystem;

/* Job object
 *  sockfd:     The stream file descritor to send the reply on.
 *  fileName:   The name of the file with data to be sent.
 */
struct Job
{
    int         sockfd;
    std::string fileName;

    Job(int sockfd, std::string const& fileName)
        : sockfd(sockfd)
        , fileName(fileName)
    {}
    ~Job()
    {
        if (sockfd)
        {
            ::close(sockfd);
        }
    }
};

class SimpleSocket
{
    int sockfd;
    public:
        SimpleSocket(int port)
        {
            sockfd = ::socket(AF_INET, SOCK_STREAM, 0);
            if (sockfd < 0)
            {
                throw std::runtime_error("Failed to create socket");
            }

            struct sockaddr_in serv_addr;
            ::bzero((char *) &serv_addr, sizeof(serv_addr));

            serv_addr.sin_family        = AF_INET;
            serv_addr.sin_addr.s_addr   = INADDR_ANY;
            serv_addr.sin_port          = htons(port);

            if (::bind(sockfd, reinterpret_cast<sockaddr*>(&serv_addr), sizeof(serv_addr)) < 0)
            {
                throw std::runtime_error("Failed to bind to port: "s + std::strerror(errno));
            }
        }
        ~SimpleSocket()
        {
            ::close(sockfd);
        }
        int waitForConnection()
        {
            int result = ::listen(sockfd,5);
            if (result < 0)
            {
                throw std::runtime_error("Listen failed");
            }
            sockaddr_in     cli_addr;
            socklen_t       clilen      = sizeof(cli_addr);
            int             newsockfd   = ::accept(sockfd, reinterpret_cast<sockaddr*>(&cli_addr), &clilen);
            if (newsockfd < 0)
            {
                throw std::runtime_error("accept failed");
            }
            return newsockfd;
        }
};

/* The set of pages the can be returned */
class Pages
{
    public:
        // Note 1: page contains the server name and page.
        // See:    getPageFromRequest()
        // Note 2: A response is preset in the files <filaName>.head <filaName>.body
        // See:    workerThread()
        static std::unique_ptr<Job> getJob(int sockfd, std::string page)
        {
            std::string  fileName = std::string("./Pages/") + page;

            bool ok = fs::is_regular_file(fileName + ".head") && fs::is_regular_file(fileName + ".body");
            if (!ok)
            {
                // Note: Invalid will not clash with saved pages.
                //       because pages are in sub-directories based on
                //       the server name.
                return std::make_unique<Job>(sockfd, "./Pages/Invalid");
            }
            return std::make_unique<Job>(sockfd, fileName);
        }
};

/* Copy a buffer to stream */
void sendToClient(int sockfd, char* buffer, int size)
{
    std::size_t write = 0;
    while(write != size)
    {
        std::size_t out = ::write(sockfd, buffer + write, size - write);
        if (out == -1 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR))
        {
            continue;
        }
        if (out == -1)
        {
            throw std::runtime_error("Failed to write to socket");
        }
        write += out;
    }
}

/* Copy a File to a stream */
void writeFile(int sockfd, std::string const& fileName)
{
    char                buffer[4096];
    std::ifstream       file(fileName.c_str());
    std::size_t         count;

    do
    {
        file.read(buffer, 4096);
        count = file.gcount();
        sendToClient(sockfd, buffer, count);
    }
    while(count > 0);
}
/* Read the request from the socket */
std::string getPageFromRequest(int connection)
{
    char buffer[4097];
    std::size_t read = 0;
    while(true)
    {
        std::size_t atual = ::read(connection, buffer + read, 4096 - read);
        if (atual == 0)
        {
            break;
        }
        if (atual == -1 && (errno == EAGAIN || errno == EINTR))
        {
            continue;
        }
        if (atual == -1)
        {
            throw std::runtime_error("Read Error");
        }
        read += atual;
        if (read == 4096)
        {
            break;
        }
    // hackery but for this simple job ¯\(0_0)/¯
        if ((read > 3) && buffer[read - 3] == '\n' && buffer[read - 2] == '\r' && buffer[read - 1] == '\n')
        {
            break;
        }
    }
    static std::string const marker = "Host: ";

    char* pageSt  = buffer + 4;
    char* pageEnd = std::find(pageSt, buffer + read, ' ');
    char* servSt  = std::search(buffer, buffer + read, marker.begin(), marker.end());
    char* servEnd = std::find(servSt + marker.size(), buffer + read, '\r');

    std::string req = std::string(servSt + marker.size(), servEnd) + std::string(pageSt, pageEnd);
    return std::string(pageSt, pageEnd);
}

void handleJob(std::unique_ptr<Job> job) {
    if (job->sockfd)
    {
        writeFile(job->sockfd, job->fileName + ".head");
        writeFile(job->sockfd, job->fileName + ".body");
    }
}

int main()
{
    std::vector<std::future<void>> pending_futures;
    SimpleSocket    socket(8100);

    int connection;
    while(connection = socket.waitForConnection())
    {
        std::string page = getPageFromRequest(connection);
        std::cout << "serving page " << page << std::endl;
        pending_futures.push_back(std::async(std::launch::async, handleJob, Pages::getJob(connection, page)));
    }

}
\$\endgroup\$
  • \$\begingroup\$ Fantastic I loved the review. It was sort of deliberate to use only C++03 features to try and get people to review it to get the C++11 -> C++17 features discussed. \$\endgroup\$ – Martin York Mar 9 '16 at 5:25
  • \$\begingroup\$ Normally I would even recommend the use of std::vector over std::list to hold Jobs in nearly every situation (because of memory locality and caching). But this may (and I emphasis may) be one situation where std::list (as you pointed out) is the better alternative. \$\endgroup\$ – Martin York Mar 9 '16 at 5:27
  • \$\begingroup\$ deprecated and will be removed in C++17 Maybe. I'll believe that when I see it. Things stay deprecated rather than removed to maintain backward compatibility. But you may be correct it is relatively easy to move to std::unique_ptr \$\endgroup\$ – Martin York Mar 9 '16 at 5:30
  • \$\begingroup\$ I had not seen the Filesystem TS thank you for that. \$\endgroup\$ – Martin York Mar 9 '16 at 5:32
  • \$\begingroup\$ auto_ptr and some other things which are superseded by better alternatives are removed by N4190 which is adopted according to Herb Sutter (secretary of the ISO C++ working group) \$\endgroup\$ – henje Mar 9 '16 at 11:39

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.