Commit 6d32e4a2 authored by octal's avatar octal

Implemented file serving

parent 7eb3e009
......@@ -57,12 +57,22 @@ class MyHandler : public Net::Http::Handler {
else if (req.resource() == "/timeout") {
timeout.arm(std::chrono::seconds(5));
}
#if 0
else if (req.resource() == "/async") {
std::thread([](Net::Http::Response response) {
std::this_thread::sleep_for(std::chrono::seconds(1));
response.send(Net::Http::Code::Ok, "Async response");
}, std::move(response)).detach();
}
#endif
else if (req.resource() == "/static") {
if (req.method() == Net::Http::Method::Get) {
Net::Http::serveFile(response, "README.md").then([](ssize_t bytes) {;
std::cout << "Sent " << bytes << " bytes" << std::endl;
}, Async::NoExcept);
}
}
}
void onTimeout(const Net::Http::Request& req, Net::Http::Response response) {
......@@ -156,8 +166,7 @@ int main(int argc, char *argv[]) {
monitor.setInterval(std::chrono::seconds(5));
monitor.start();
server->serveThreaded();
std::this_thread::sleep_for(std::chrono::seconds(10));
server->serve();
std::cout << "Shutdowning server" << std::endl;
server->shutdown();
monitor.shutdown();
......
......@@ -9,6 +9,10 @@
#include <stdexcept>
#include <ctime>
#include <iomanip>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include "common.h"
#include "http.h"
#include "net.h"
......@@ -21,7 +25,7 @@ namespace Net {
namespace Http {
template<typename H, typename Stream, typename... Args>
typename std::enable_if<Header::IsHeader<H>::value, void>::type
typename std::enable_if<Header::IsHeader<H>::value, Stream&>::type
writeHeader(Stream& stream, Args&& ...args) {
H header(std::forward<Args>(args)...);
......@@ -29,10 +33,55 @@ writeHeader(Stream& stream, Args&& ...args) {
header.write(stream);
stream << crlf;
return stream;
}
static constexpr const char* ParserData = "__Parser";
namespace {
bool writeStatusLine(Code code, DynamicStreamBuf& buf) {
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) return false; \
} while (0)
std::ostream os(&buf);
OUT(os << "HTTP/1.1 ");
OUT(os << static_cast<int>(code));
OUT(os << ' ');
OUT(os << code);
OUT(os << crlf);
return true;
#undef OUT
}
bool writeHeaders(const Header::Collection& headers, DynamicStreamBuf& buf) {
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) return false; \
} while (0)
std::ostream os(&buf);
for (const auto& header: headers.list()) {
OUT(os << header->name() << ": ");
OUT(header->write(os));
OUT(os << crlf);
}
return true;
#undef OUT
}
}
namespace Private {
void
......@@ -366,57 +415,37 @@ Request::peer() const {
}
#endif
void
ResponseStream::writeStatusLine() {
std::ostream os(&buf_);
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) { \
throw Error("Response exceeded buffer size"); \
} \
} while (0);
OUT(os << "HTTP/1.1 ");
OUT(os << static_cast<int>(code_));
OUT(os << ' ');
OUT(os << code_);
OUT(os << crlf);
#undef OUT
}
ResponseStream::ResponseStream(
Message&& other,
std::weak_ptr<Tcp::Peer> peer,
Tcp::IoWorker* io,
size_t streamSize)
: Message(std::move(other))
, peer_(std::move(peer))
, buf_(streamSize)
, io_(io)
{
if (!writeStatusLine(code_, buf_))
throw Error("Response exceeded buffer size");
void
ResponseStream::writeHeaders() {
if (writeHeaders(headers_, buf_)) {
std::ostream os(&buf_);
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) { \
throw Error("Response exceeded buffer size"); \
} \
} while (0);
for (const auto& header: headers_.list()) {
OUT(os << header->name() << ": ");
OUT(header->write(os));
OUT(os << crlf);
if (!writeHeader<Header::TransferEncoding>(os, Header::Encoding::Chunked))
throw Error("Response exceeded buffer size");
os << crlf;
}
else {
throw Error("Response exceeded buffer size");
}
OUT(writeHeader<Header::TransferEncoding>(os, Header::Encoding::Chunked));
OUT(os << crlf);
#undef OUT
}
void
ResponseStream::flush() {
io_->disarmTimer();
auto buf = buf_.buffer();
peer()->send(buf);
auto fd = peer()->fd();
io_->asyncWrite(fd, buf);
buf_.clear();
}
......@@ -448,17 +477,8 @@ Response::putOnWire(const char* data, size_t len)
} \
} while (0);
OUT(os << "HTTP/1.1 ");
OUT(os << static_cast<int>(code_));
OUT(os << ' ');
OUT(os << code_);
OUT(os << crlf);
for (const auto& header: headers_.list()) {
OUT(os << header->name() << ": ");
OUT(header->write(os));
OUT(os << crlf);
}
OUT(writeStatusLine(code_, buf_));
OUT(writeHeaders(headers_, buf_));
OUT(writeHeader<Header::ContentLength>(os, len));
......@@ -474,13 +494,82 @@ Response::putOnWire(const char* data, size_t len)
#undef OUT
return peer()->send(buffer);
auto fd = peer()->fd();
return io_->asyncWrite(fd, buffer);
} catch (const std::runtime_error& e) {
return Async::Promise<ssize_t>::rejected(e);
}
}
Async::Promise<ssize_t>
serveFile(Response& response, const char* fileName, const Mime::MediaType& contentType)
{
struct stat sb;
int fd = open(fileName, O_RDONLY);
if (fd == -1) {
/* @Improvement: maybe could we check for errno here and emit a different error
message
*/
throw HttpError(Net::Http::Code::Not_Found, "");
}
int res = ::fstat(fd, &sb);
if (res == -1) {
throw HttpError(Code::Internal_Server_Error, "");
}
auto *buf = response.rdbuf();
std::ostream os(buf);
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) { \
return Async::Promise<ssize_t>::rejected(Error("Response exceeded buffer size")); \
} \
} while (0);
auto setContentType = [&](const Mime::MediaType& contentType) {
auto& headers = response.headers();
auto ct = headers.tryGet<Header::ContentType>();
if (ct)
ct->setMime(contentType);
else
headers.add<Header::ContentType>(contentType);
};
OUT(writeStatusLine(Http::Code::Ok, *buf));
if (contentType.isValid()) {
setContentType(contentType);
} else {
auto mime = Mime::MediaType::fromFile(fileName);
if (mime.isValid())
setContentType(mime);
}
OUT(writeHeaders(response.headers(), *buf));
const size_t len = sb.st_size;
OUT(writeHeader<Header::ContentLength>(os, len));
OUT(os << crlf);
auto *io = response.io_;
auto peer = response.peer();
auto sockFd = peer->fd();
auto buffer = buf->buffer();
return io->asyncWrite(sockFd, buffer, MSG_MORE).then([=](ssize_t bytes) {
return io->asyncWrite(sockFd, FileBuffer(fileName));
}, Async::Throw);
#undef OUT
}
void
Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer) {
try {
......
......@@ -220,18 +220,7 @@ private:
Message&& other,
std::weak_ptr<Tcp::Peer> peer,
Tcp::IoWorker* io,
size_t streamSize)
: Message(std::move(other))
, peer_(std::move(peer))
, buf_(streamSize)
, io_(io)
{
writeStatusLine();
writeHeaders();
}
void writeStatusLine();
void writeHeaders();
size_t streamSize);
std::shared_ptr<Tcp::Peer> peer() const {
if (peer_.expired())
......@@ -277,6 +266,8 @@ public:
friend class Handler;
friend class Timeout;
friend Async::Promise<ssize_t> serveFile(Response&, const char *, const Mime::MediaType&);
static constexpr size_t DefaultStreamSize = 512;
// C++11: std::weak_ptr move constructor is C++14 only so the default
......@@ -364,6 +355,15 @@ public:
return ResponseStream(std::move(*this), peer_, io_, streamSize);
}
// Unsafe API
DynamicStreamBuf *rdbuf() {
return &buf_;
}
DynamicStreamBuf *rdbuf(DynamicStreamBuf* other) {
throw std::domain_error("Unimplemented");
}
private:
Response(Tcp::IoWorker* io)
......@@ -394,6 +394,9 @@ private:
Tcp::IoWorker *io_;
};
Async::Promise<ssize_t> serveFile(
Response& response, const char *fileName,
const Mime::MediaType& contentType = Mime::MediaType());
namespace Private {
......
......@@ -10,6 +10,7 @@
#include "peer.h"
#include "os.h"
#include <sys/timerfd.h>
#include <sys/sendfile.h>
namespace Net {
......@@ -343,66 +344,49 @@ IoWorker::handleWriteQueue() {
if (!entry) break;
const auto &write = entry->data();
asyncWriteImpl(write.fd, write);
asyncWriteImpl(write.peerFd, write);
}
}
Async::Promise<ssize_t>
IoWorker::asyncWrite(Fd fd, const Buffer& buffer) {
// If the I/O operation has been initiated from an other thread, we queue it and we'll process
// it in our own thread so that we make sure that every I/O operation happens in the right thread
if (std::this_thread::get_id() != thisId) {
return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
auto detached = buffer.detach();
OnHoldWrite write(std::move(resolve), std::move(reject), detached);
write.fd = fd;
auto *e = writesQueue.allocEntry(write);
writesQueue.push(e);
});
}
return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
auto it = toWrite.find(fd);
if (it != std::end(toWrite)) {
reject(Net::Error("Multiple writes on the same fd"));
return;
}
asyncWriteImpl(fd, buffer, resolve, reject);
});
}
void
IoWorker::asyncWriteImpl(Fd fd, const IoWorker::OnHoldWrite& entry, WriteStatus status) {
asyncWriteImpl(fd, entry.buffer, entry.resolve, entry.reject, status);
asyncWriteImpl(fd, entry.flags, entry.buffer, entry.resolve, entry.reject, status);
}
void
IoWorker::asyncWriteImpl(
Fd fd, const Buffer& buffer,
Fd fd, int flags, const BufferHolder& buffer,
Async::Resolver resolve, Async::Rejection reject, WriteStatus status)
{
auto cleanUp = [&]() {
if (buffer.isOwned) delete[] buffer.data;
if (buffer.isRaw()) {
auto raw = buffer.raw();
if (raw.isOwned) delete[] raw.data;
}
if (status == Retry)
toWrite.erase(fd);
};
ssize_t totalWritten = 0;
for (;;) {
auto *ptr = buffer.data + totalWritten;
auto len = buffer.len - totalWritten;
ssize_t bytesWritten = ::send(fd, ptr, len, 0);
ssize_t bytesWritten = 0;
auto len = buffer.size() - totalWritten;
if (buffer.isRaw()) {
auto raw = buffer.raw();
auto ptr = raw.data + totalWritten;
bytesWritten = ::send(fd, ptr, len, flags);
} else {
auto file = buffer.fd();
off_t offset = totalWritten;
bytesWritten = ::sendfile(fd, file, &offset, len);
}
if (bytesWritten < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
auto buf = buffer.isOwned ? buffer : buffer.detach(totalWritten);
if (status == FirstTry) {
toWrite.insert(
std::make_pair(fd,
OnHoldWrite(std::move(resolve), std::move(reject), buf)));
OnHoldWrite(std::move(resolve), std::move(reject), buffer.detach(totalWritten), flags)));
}
poller.rearmFd(fd, NotifyOn::Read | NotifyOn::Write, Polling::Tag(fd), Polling::Mode::Edge);
}
......@@ -423,8 +407,6 @@ IoWorker::asyncWriteImpl(
}
}
} // namespace Tcp
} // namespace Net
......@@ -12,6 +12,7 @@
#include "tcp.h"
#include "async.h"
#include "stream.h"
#include "net.h"
#include <thread>
#include <mutex>
......@@ -59,22 +60,34 @@ public:
});
}
private:
struct OnHoldWrite {
OnHoldWrite(Async::Resolver resolve, Async::Rejection reject,
Buffer buffer)
: resolve(std::move(resolve))
, reject(std::move(reject))
, buffer(std::move(buffer))
, fd(-1)
{ }
template<typename Buf>
Async::Promise<ssize_t> asyncWrite(Fd fd, const Buf& buffer, int flags = 0) {
// If the I/O operation has been initiated from an other thread, we queue it and we'll process
// it in our own thread so that we make sure that every I/O operation happens in the right thread
if (std::this_thread::get_id() != thisId) {
return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
BufferHolder holder(buffer);
auto detached = holder.detach();
OnHoldWrite write(std::move(resolve), std::move(reject), detached, flags);
write.peerFd = fd;
auto *e = writesQueue.allocEntry(write);
writesQueue.push(e);
});
}
return Async::Promise<ssize_t>([&](Async::Resolver& resolve, Async::Rejection& reject) {
Async::Resolver resolve;
Async::Rejection reject;
auto it = toWrite.find(fd);
if (it != std::end(toWrite)) {
reject(Net::Error("Multiple writes on the same fd"));
return;
}
Buffer buffer;
Fd fd;
};
asyncWriteImpl(fd, flags, BufferHolder(buffer), resolve, reject);
});
}
private:
void
armTimerMs(std::chrono::milliseconds value, Async::Resolver, Async::Rejection reject);
......@@ -110,10 +123,98 @@ private:
Retry
};
struct BufferHolder {
enum Type { Raw, File };
explicit BufferHolder(const Buffer& buffer)
: type(Raw)
, u(buffer)
{
size_ = buffer.len;
}
explicit BufferHolder(const FileBuffer& buffer)
: type(File)
, u(buffer.fd())
{
size_ = buffer.size();
}
bool isFile() const { return type == File; }
bool isRaw() const { return type == Raw; }
size_t size() const { return size_; }
Fd fd() const {
if (!isFile())
throw std::runtime_error("Tried to retrieve fd of a non-filebuffer");
return u.fd;
}
Buffer raw() const {
if (!isRaw())
throw std::runtime_error("Tried to retrieve raw data of a non-buffer");
return u.raw;
}
BufferHolder detach(size_t offset = 0) const {
if (!isRaw())
return BufferHolder(u.fd, size_);
if (u.raw.isOwned)
return BufferHolder(u.raw);
auto detached = u.raw.detach(offset);
return BufferHolder(detached);
}
private:
BufferHolder(Fd fd, size_t size)
: u(fd)
, size_(size)
, type(File)
{ }
union U {
Buffer raw;
Fd fd;
U(Buffer buffer) : raw(buffer) { }
U(Fd fd) : fd(fd) { }
} u;
size_t size_;
Type type;
};
struct OnHoldWrite {
OnHoldWrite(Async::Resolver resolve, Async::Rejection reject,
BufferHolder buffer, int flags = 0)
: resolve(std::move(resolve))
, reject(std::move(reject))
, buffer(std::move(buffer))
, flags(flags)
, peerFd(-1)
{ }
Async::Resolver resolve;
Async::Rejection reject;
BufferHolder buffer;
int flags;
Fd peerFd;
};
Polling::Epoll poller;
std::unique_ptr<std::thread> thread;
mutable std::mutex peersMutex;
std::unordered_map<Fd, std::shared_ptr<Peer>> peers;
/* @Incomplete: this should be a std::dequeue.
If an asyncWrite on a particular fd is initiated whereas the fd is not write-ready
yet and some writes are still on-hold, writes should queue-up so that when the
fd becomes ready again, we can write everything
*/
std::unordered_map<Fd, OnHoldWrite> toWrite;
Optional<Timer> timer;
......@@ -133,11 +234,9 @@ private:
std::shared_ptr<Peer>& getPeer(Fd fd);
std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
Async::Promise<ssize_t> asyncWrite(Fd fd, const Buffer& buffer);
void asyncWriteImpl(Fd fd, const OnHoldWrite& entry, WriteStatus status = FirstTry);
void asyncWriteImpl(
Fd fd, const Buffer& buffer,
Fd fd, int flags, const BufferHolder& buffer,
Async::Resolver resolve, Async::Rejection reject,
WriteStatus status = FirstTry);
......
......@@ -179,7 +179,7 @@ Listener::bind(const Address& address) {
}
make_non_blocking(fd);
poller.addFd(fd, Polling::NotifyOn::Write, Polling::Tag(fd), Polling::Mode::Edge);
poller.addFd(fd, Polling::NotifyOn::Read, Polling::Tag(fd), Polling::Mode::Edge);
listen_fd = fd;
g_listen_fd = fd;
......@@ -210,7 +210,7 @@ Listener::run() {
if (event.tag == shutdownFd.tag())
return;
else {
if (event.flags.hasFlag(Polling::NotifyOn::Write)) {
if (event.flags.hasFlag(Polling::NotifyOn::Read)) {
auto fd = event.tag.value();
if (fd == listen_fd)
handleNewConnection();
......
......@@ -51,6 +51,50 @@ MediaType::fromRaw(const char* str, size_t len) {
return res;
}
MediaType
MediaType::fromFile(const char* fileName)
{
const char *extensionOffset = nullptr;
const char *p = fileName;
while (*p) {
if (*p == '.')
extensionOffset = p;
++p;
}
if (!extensionOffset)
return MediaType();
++extensionOffset;
struct Extension {
const char* const raw;
Mime::Type top;
Mime::Subtype sub;
};
// @Data: maybe one day try to export http://www.iana.org/assignments/media-types/media-types.xhtml
// as an item-list
static constexpr Extension KnownExtensions[] = {
{ "jpg", Type::Image, Subtype::Jpeg },
{ "jpeg", Type::Image, Subtype::Jpeg },
{ "png", Type::Image, Subtype::Png },
{ "bmp", Type::Image, Subtype::Bmp },
{ "txt", Type::Text, Subtype::Plain },
{ "md", Type::Text, Subtype::Plain }
};
for (const auto& ext: KnownExtensions) {
if (!strcmp(extensionOffset, ext.raw)) {
return MediaType(ext.top, ext.sub);
}
}
return MediaType();
}
void
MediaType::parseRaw(const char* str, size_t len) {
auto raise = [&](const char* str) {
......
......@@ -156,6 +156,8 @@ public:
static MediaType fromString(const std::string& str);
static MediaType fromString(std::string&& str);
static MediaType fromFile(const char* fileName);
Mime::Type top() const { return top_; }
Mime::Subtype sub() const { return sub_; }
Mime::Suffix suffix() const { return suffix_; }
......
......@@ -78,11 +78,10 @@ Peer::tryGetData(std::string(name)) const {
}
Async::Promise<ssize_t>
Peer::send(const Buffer& buffer) {
return io_->asyncWrite(fd_, buffer);
Peer::send(const Buffer& buffer, int flags) {
return io_->asyncWrite(fd_, buffer, flags);
}
std::ostream& operator<<(std::ostream& os, const Peer& peer) {
const auto& addr = peer.address();
os << "(" << addr.host() << ", " << addr.port() << ") [" << peer.hostname() << "]";
......
......@@ -51,7 +51,7 @@ public:
return std::static_pointer_cast<T>(data);
}
Async::Promise<ssize_t> send(const Buffer& buffer);
Async::Promise<ssize_t> send(const Buffer& buffer, int flags = 0);
private:
IoWorker* io_;
......
......@@ -6,6 +6,40 @@
#include "stream.h"
#include <algorithm>
#include <iostream>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
FileBuffer::FileBuffer(const char* fileName)
: fileName_(fileName)
{
init(fileName);
}
FileBuffer::FileBuffer(const std::string& fileName)
: fileName_(fileName)
{
init(fileName.c_str());
}
void
FileBuffer::init(const char* fileName)
{
int fd = open(fileName, O_RDONLY);
if (fd == -1) {
throw std::runtime_error("Could not open file");
}
struct stat sb;
int res = ::fstat(fd, &sb);
if (res == -1) {
throw std::runtime_error("Could not get file stats");
}
fd_ = fd;
size_ = sb.st_size;
}
DynamicStreamBuf::int_type
DynamicStreamBuf::overflow(DynamicStreamBuf::int_type ch) {
......
......@@ -13,6 +13,7 @@
#include <vector>
#include <limits>
#include <iostream>
#include "os.h"
static constexpr char CR = 0xD;
static constexpr char LF = 0xA;
......@@ -115,6 +116,12 @@ private:
};
struct Buffer {
Buffer()
: data(nullptr)
, len(0)
, isOwned(false)
{ }
Buffer(const char * const data, size_t len, bool own = false)
: data(data)
, len(len)
......@@ -136,6 +143,23 @@ struct Buffer {
const bool isOwned;
};
struct FileBuffer {
FileBuffer() { }
FileBuffer(const char* fileName);
FileBuffer(const std::string& fileName);
std::string fileName() const { return fileName_; }
Fd fd() const { return fd_; }
size_t size() const { return size_; }
private:
void init(const char* fileName);
std::string fileName_;
Fd fd_;
size_t size_;
};
class DynamicStreamBuf : public StreamBuf<char> {
public:
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment