Unverified Commit 9c670280 authored by Dennis Jenkins's avatar Dennis Jenkins Committed by GitHub

Merge pull request #430 from AdrianCX/master

Fix issues in test_memcheck
parents 98ce7204 c1cfaefc
......@@ -171,13 +171,15 @@ namespace Async {
struct Core {
Core(State _state, TypeId _id)
: state(_state)
: allocated(false)
, state(_state)
, exc()
, mtx()
, requests()
, id(_id)
{ }
bool allocated;
State state;
std::exception_ptr exc;
......@@ -210,11 +212,18 @@ namespace Async {
}
void *mem = memory();
if (allocated) {
reinterpret_cast<T*>(mem)->~T();
allocated = false;
}
new (mem) T(std::forward<Args>(args)...);
allocated = true;
state = State::Fulfilled;
}
virtual ~Core() { }
virtual ~Core() {}
};
template<typename T>
......@@ -224,14 +233,18 @@ namespace Async {
, storage()
{ }
~CoreT() {
if (allocated) {
reinterpret_cast<T*>(&storage)->~T();
allocated = false;
}
}
template<class Other>
struct Rebind {
typedef CoreT<Other> Type;
};
typedef typename std::aligned_storage<sizeof(T), alignof(T)>::type Storage;
Storage storage;
T& value() {
if (state != State::Fulfilled)
throw Error("Attempted to take the value of a not fulfilled promise");
......@@ -241,9 +254,14 @@ namespace Async {
bool isVoid() const override { return false; }
protected:
void *memory() override {
return &storage;
}
private:
typedef typename std::aligned_storage<sizeof(T), alignof(T)>::type Storage;
Storage storage;
};
template<>
......@@ -254,6 +272,7 @@ namespace Async {
bool isVoid() const override { return true; }
protected:
void *memory() override {
return nullptr;
}
......
......@@ -205,18 +205,21 @@ private:
struct ConnectionEntry {
ConnectionEntry(
Async::Resolver resolve, Async::Rejection reject,
std::shared_ptr<Connection> connection, const struct sockaddr* addr, socklen_t addr_len)
std::shared_ptr<Connection> connection, const struct sockaddr* _addr, socklen_t _addr_len)
: resolve(std::move(resolve))
, reject(std::move(reject))
, connection(connection)
, addr(addr)
, addr_len(addr_len)
{ }
{
addr_len = _addr_len;
memcpy(&addr, _addr, addr_len);
}
const sockaddr *getAddr() { return reinterpret_cast<const sockaddr *>(&addr); }
Async::Resolver resolve;
Async::Rejection reject;
std::weak_ptr<Connection> connection;
const struct sockaddr* addr;
sockaddr_storage addr;
socklen_t addr_len;
};
......
......@@ -168,7 +168,7 @@ public:
Optional<T> &operator=(Optional<T> &&other)
noexcept(types::is_nothrow_move_constructible<T>::value)
{
if (other.data()) {
if (!other.isEmpty()) {
move_helper(std::move(other), types::is_move_constructible<T>());
other.none_flag = NoneMarker;
}
......
......@@ -7,6 +7,7 @@
#include <pistache/client.h>
#include <pistache/http.h>
#include <pistache/stream.h>
#include <pistache/net.h>
#include <sys/sendfile.h>
#include <netdb.h>
......@@ -283,7 +284,7 @@ Transport::handleConnectionQueue() {
if (!conn) {
throw std::runtime_error("Connection error");
}
int res = ::connect(conn->fd, data->addr, data->addr_len);
int res = ::connect(conn->fd, data->getAddr(), data->addr_len);
if (res == -1) {
if (errno == EINPROGRESS) {
reactor()->registerFdOneShot(key(), conn->fd, NotifyOn::Write | NotifyOn::Hangup | NotifyOn::Shutdown);
......@@ -358,7 +359,6 @@ void
Connection::connect(const Address& addr)
{
struct addrinfo hints;
struct addrinfo *addrs;
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = addr.family();
hints.ai_socktype = SOCK_STREAM; /* Stream socket */
......@@ -367,11 +367,15 @@ Connection::connect(const Address& addr)
const auto& host = addr.host();
const auto& port = addr.port().toString();
TRY(::getaddrinfo(host.c_str(), port.c_str(), &hints, &addrs));
AddrInfo addressInfo;
TRY(addressInfo.invoke(host.c_str(), port.c_str(), &hints));
const addrinfo *addrs = addressInfo.get_info_ptr();
int sfd = -1;
for (struct addrinfo *addr = addrs; addr; addr = addr->ai_next) {
for (const addrinfo *addr = addrs; addr; addr = addr->ai_next) {
sfd = ::socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
if (sfd < 0) continue;
......
......@@ -238,9 +238,7 @@ Address::init(const std::string& addr) {
family_ = AF_INET6;
try {
in6_addr addr6;
char buff6[INET6_ADDRSTRLEN + 1] = {0, };
std::copy(host_.begin(), host_.end(), buff6);
inet_pton(AF_INET6, buff6, &(addr6.s6_addr16));
inet_pton(AF_INET6, host_.c_str(), &(addr6.s6_addr16));
} catch (std::runtime_error) {
throw std::invalid_argument("Invalid IPv6 address");
}
......@@ -257,9 +255,7 @@ Address::init(const std::string& addr) {
}
try {
in_addr addr;
char buff[INET_ADDRSTRLEN + 1] = {0, };
std::copy(host_.begin(), host_.end(), buff);
inet_pton(AF_INET, buff, &(addr));
inet_pton(AF_INET, host_.c_str(), &(addr));
} catch (std::runtime_error) {
throw std::invalid_argument("Invalid IPv4 address");
}
......
......@@ -270,8 +270,12 @@ Transport::asyncWriteImpl(Fd fd)
}
if (bytesWritten < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
auto bufferHolder = buffer.detach(totalWritten);
// pop_front kills buffer - so we cannot continue loop or use buffer after this point
wq.pop_front();
wq.push_front(WriteEntry(std::move(deferred), buffer.detach(totalWritten), flags));
wq.push_front(WriteEntry(std::move(deferred), bufferHolder, flags));
reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
}
else {
......
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