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));
}
}
