diff --git a/common/cpp/include/system_wrappers/system_io.h b/common/cpp/include/system_wrappers/system_io.h
index ed2cce30801c44b191b9091f96ef5c3afa677cbc..3688ad7208494e01a277d4621e95b3c0c1580e46 100644
--- a/common/cpp/include/system_wrappers/system_io.h
+++ b/common/cpp/include/system_wrappers/system_io.h
@@ -22,7 +22,7 @@ class SystemIO final : public IO {
// Function maps. Should never be called apart from in wrapper function
FileDescriptor _open(const char* filename, int posix_open_flags) const;
- void _close(FileDescriptor fd) const;
+ bool _close(FileDescriptor fd) const;
ssize_t _read(FileDescriptor fd, void* buffer, size_t length) const;
ssize_t _write(FileDescriptor fd, const void* buffer, size_t count) const;
int _mkdir(const char* dirname) const;
@@ -33,7 +33,7 @@ class SystemIO final : public IO {
SocketDescriptor _accept(SocketDescriptor socket_fd, void* address, size_t* address_length) const;
ssize_t _send(SocketDescriptor socket_fd, const void* buffer, size_t length) const;
ssize_t _recv(SocketDescriptor socket_fd, void* buffer, size_t length) const;
- void _close_socket(SocketDescriptor socket_fd) const;
+ bool _close_socket(SocketDescriptor socket_fd) const;
std::unique_ptr<std::tuple<std::string, uint16_t>> SplitAddressToHostAndPort(std::string address) const;
diff --git a/common/cpp/src/system_io.cpp b/common/cpp/src/system_io.cpp
index ab3b0eaf3b776a40745b3065b37ca2078edee7fc..89de7fb44912ea9c62e7300ec4c96bbab4064047 100644
--- a/common/cpp/src/system_io.cpp
+++ b/common/cpp/src/system_io.cpp
@@ -110,20 +110,26 @@ hidra2::FileDescriptor hidra2::SystemIO::Open(const std::string& filename,
IOErrors* err) const {
int flags = FileOpenModeToPosixFileOpenMode(open_flags);
FileDescriptor fd = _open(filename.c_str(), flags);
- *err = GetLastError();
+ if(fd == -1) {
+ *err = GetLastError();
+ }
return fd;
}
void hidra2::SystemIO::CloseSocket(SocketDescriptor fd, hidra2::IOErrors* err) const {
- _close_socket(fd);
- if (err) {
+ if(err) {
+ *err = IOErrors::kNoError;
+ }
+ if(!_close_socket(fd) && err) {
*err = GetLastError();
}
}
void hidra2::SystemIO::Close(FileDescriptor fd, hidra2::IOErrors* err) const {
- _close(fd);
- if (err) {
+ if(err) {
+ *err = IOErrors::kNoError;
+ }
+ if(!_close(fd) && err) {
*err = GetLastError();
}
}
@@ -134,16 +140,18 @@ size_t hidra2::SystemIO::Read(FileDescriptor fd, void* buf, size_t length, IOErr
size_t already_read = 0;
while(already_read < length) {
- ssize_t received_amount = _read(fd, (uint8_t*)buf + already_read, length - already_read);
- if(received_amount == 0) {
+ ssize_t read_amount = _read(fd, (uint8_t*)buf + already_read, length - already_read);
+ if(read_amount == 0) {
*err = IOErrors::kEndOfFile;
return already_read;
}
- *err = GetLastError();
- if (*err != IOErrors::kNoError) {
- return already_read;
+ if (read_amount == -1) {
+ *err = GetLastError();
+ if (*err != IOErrors::kNoError) {
+ return already_read;
+ }
}
- already_read += received_amount;
+ already_read += read_amount;
}
return already_read;
@@ -155,16 +163,18 @@ size_t hidra2::SystemIO::Write(FileDescriptor fd, const void* buf, size_t length
size_t already_wrote = 0;
while(already_wrote < length) {
- ssize_t send_amount = _write(fd, (uint8_t*)buf + already_wrote, length - already_wrote);
- if(send_amount == 0) {
+ ssize_t write_amount = _write(fd, (uint8_t*)buf + already_wrote, length - already_wrote);
+ if(write_amount == 0) {
*err = IOErrors::kEndOfFile;
return already_wrote;
}
- *err = GetLastError();
- if (*err != IOErrors::kNoError) {
- return already_wrote;
+ if (write_amount == -1) {
+ *err = GetLastError();
+ if (*err != IOErrors::kNoError) {
+ return already_wrote;
+ }
}
- already_wrote += send_amount;
+ already_wrote += write_amount;
}
return already_wrote;
@@ -272,9 +282,11 @@ size_t hidra2::SystemIO::Receive(SocketDescriptor socket_fd, void* buf, size_t l
*err = IOErrors::kEndOfFile;
return already_received;
}
- *err = GetLastError();
- if (*err != IOErrors::kNoError) {
- return already_received;
+ if (received_amount == -1) {
+ *err = GetLastError();
+ if (*err != IOErrors::kNoError) {
+ return already_received;
+ }
}
already_received += received_amount;
}
@@ -321,9 +333,11 @@ size_t hidra2::SystemIO::Send(SocketDescriptor socket_fd,
*err = IOErrors::kEndOfFile;
return already_sent;
}
- *err = GetLastError();
- if (*err != IOErrors::kNoError) {
- return already_sent;
+ if (send_amount == -1) {
+ *err = GetLastError();
+ if (*err != IOErrors::kNoError) {
+ return already_sent;
+ }
}
already_sent += send_amount;
}
diff --git a/common/cpp/src/system_io_linux.cpp b/common/cpp/src/system_io_linux.cpp
index 42e718cdda911a673596750345861be7ebf14cac..9ac0c1655e9571def4128b10403f660fa2e64af2 100644
--- a/common/cpp/src/system_io_linux.cpp
+++ b/common/cpp/src/system_io_linux.cpp
@@ -162,8 +162,8 @@ hidra2::FileDescriptor hidra2::SystemIO::_open(const char* filename, int posix_o
return ::open(filename, posix_open_flags, S_IWUSR | S_IRWXU);
}
-void SystemIO::_close(hidra2::FileDescriptor fd) const {
- ::close(fd);
+bool SystemIO::_close(hidra2::FileDescriptor fd) const {
+ return ::close(fd) == 0;
}
ssize_t SystemIO::_read(hidra2::FileDescriptor fd, void* buffer, size_t length) const {
@@ -202,8 +202,8 @@ SocketDescriptor SystemIO::_accept(SocketDescriptor socket_fd, void* address, si
return ::accept(socket_fd, static_cast<sockaddr*>(address), reinterpret_cast<socklen_t*>(address_length));
}
-void SystemIO::_close_socket(SocketDescriptor socket_fd) const {
- ::close(socket_fd);
+bool SystemIO::_close_socket(SocketDescriptor socket_fd) const {
+ return ::close(socket_fd) == 0;
}
}
diff --git a/examples/producer/dummy-data-producer/dummy_data_producer.cpp b/examples/producer/dummy-data-producer/dummy_data_producer.cpp
index f740beada136c2267b397accc529c5bc13129ee3..5dbad173de1f182c01f33e10b3b9b1058504ad76 100644
--- a/examples/producer/dummy-data-producer/dummy_data_producer.cpp
+++ b/examples/producer/dummy-data-producer/dummy_data_producer.cpp
@@ -33,7 +33,7 @@ int SendDummyData(const std::string& receiver_address, size_t number_of_byte, ui
auto buffer = std::unique_ptr<uint8_t>(new uint8_t[number_of_byte]);
- for(uint64_t i = 1; i < iterations; i++) {
+ for(uint64_t i = 0; i < iterations; i++) {
std::cout << "Send file " << i + 1 << "/" << iterations << std::endl;
hidra2::ProducerError error;
error = producer->Send(i, buffer.get(), number_of_byte);
diff --git a/receiver/src/network_producer_peer.cpp b/receiver/src/network_producer_peer.cpp
index 528c72d4abe1dc6408571686724f85f0e3ecb501..41b5e7c8c9b9a26f44442721f950f52f71ba23d9 100644
--- a/receiver/src/network_producer_peer.cpp
+++ b/receiver/src/network_producer_peer.cpp
@@ -116,9 +116,12 @@ size_t NetworkProducerPeer::handle_generic_request_(GenericNetworkRequest* reque
assert(handler_information.response_size <= kGenericBufferSize);//Would overwrite arbitrary memory
IOErrors err;
+
+ static const size_t sizeof_generic_request = sizeof(GenericNetworkRequest);
//receive the rest of the message
- io->Receive(socket_fd_, request + sizeof(GenericNetworkRequest),
- handler_information.request_size - sizeof(GenericNetworkRequest), &err);
+ size_t rec = io->Receive(socket_fd_, (uint8_t*)request + sizeof_generic_request,
+ handler_information.request_size - sizeof_generic_request, &err);
+ std::cout << "rec:" << rec << std::endl;
if(err != IOErrors::kNoError) {
std::cerr << "[" << connection_id() << "] NetworkProducerPeer::handle_generic_request_/receive_timeout: " <<
request->op_code << std::endl;
diff --git a/receiver/src/network_producer_peer_handlers.cpp b/receiver/src/network_producer_peer_handlers.cpp
index 2ad818293fc979d3fe441fe2e72d350ad53791d1..212f72c1b418279dcb8699d41429ae8344f154af 100644
--- a/receiver/src/network_producer_peer_handlers.cpp
+++ b/receiver/src/network_producer_peer_handlers.cpp
@@ -20,7 +20,7 @@ const std::vector<NetworkProducerPeer::RequestHandlerInformation> NetworkProduce
void NetworkProducerPeer::handle_send_data_request_(NetworkProducerPeer* self, const SendDataRequest* request,
SendDataResponse* response) {
- IOErrors ioErr;
+ IOErrors io_err;
if (request->file_size == 0) {
std::cerr << "[" << self->connection_id() << "] file_id: " << request->file_id << " has size of 0!" << std::endl;
@@ -28,17 +28,17 @@ void NetworkProducerPeer::handle_send_data_request_(NetworkProducerPeer* self, c
return;
}
- if(request->file_size > size_t(2)*size_t(1024)*size_t(1024)*size_t(1024)/*2GiByte*/) {
+ if(request->file_size > size_t(1024)*size_t(1024)*size_t(1024)*size_t(2)/*2GiByte*/) {
response->error_code = NET_ERR__ALLOCATE_STORAGE_FAILED;
return;
}
- FileDescriptor fd = self->CreateAndOpenFileByFileId(request->file_id, &ioErr);
- if(ioErr != IOErrors::kNoError) {
+ FileDescriptor fd = self->CreateAndOpenFileByFileId(request->file_id, &io_err);
+ if(io_err != IOErrors::kNoError) {
response->error_code = NET_ERR__FILENAME_ALREADY_IN_USE;
std::cerr << "[" << self->connection_id() << "] file_id: " << request->file_id << " does already exists" << std::endl;
- self->io->Skip(self->socket_fd_, request->file_size, &ioErr);
- if(ioErr != IOErrors::kNoError) {
+ self->io->Skip(self->socket_fd_, request->file_size, &io_err);
+ if(io_err != IOErrors::kNoError) {
std::cout << "[NetworkProducerPeer] Out of sync force disconnect" << std::endl;
self->io->CloseSocket(self->socket_fd_, nullptr);
}
@@ -56,13 +56,13 @@ void NetworkProducerPeer::handle_send_data_request_(NetworkProducerPeer* self, c
return;
}
- self->io->Receive(self->socket_fd_, buffer.get(), request->file_size, &ioErr);
- if(ioErr != IOErrors::kNoError) {
+ self->io->Receive(self->socket_fd_, buffer.get(), request->file_size, &io_err);
+ if(io_err != IOErrors::kNoError) {
std::cerr << "[" << self->connection_id() << "] An IO error occurred while receiving the file" << std::endl;
response->error_code = NET_ERR__INTERNAL_SERVER_ERROR;
}
- self->io->Write(fd, buffer.get(), request->file_size, &ioErr);
+ self->io->Write(fd, buffer.get(), request->file_size, &io_err);
response->error_code = NET_ERR__NO_ERROR;
}