4
\$\begingroup\$

Please help me establish a solid foundation for this computational neuroscience project. I'd like to make sure I'm moving forward with something reasonable before I begin tackling the more complex parts involving the inclusion of a linear algebra library and algorithms for decoding/interpreting incoming streams of neuron spike rasters.

There are 3 main pieces of functionality that could use some review.

  1. network protocol agnostic data receive and deserialization
  2. epoll-based data read on a TCP socket
  3. deserialization from unsigned char buffer to a vector of uint64_t (event times)

(If you have time to provide stylistic critiques on git maintenance, idiomatic language usage, etc. that's great too.)

Below is pretty much the meat of it, but here is the repo for those who would like some more context: https://github.com/andresito00/gray-decoders/tree/master

  1. Agnostic rx/deserialization. I don't particularly like the constant resizing, but a few posts I've read seem to indicate this is a safer way to go about it. WDYT? The Receiver class containing this method is templated on Queue type (Q), Data Element Type (S), and Network Stack (T net_xport_). The following function is responsible for invoking items #2 and #3.
  ReceiverStatus receive(Q& q)
  {
    status_ = ReceiverStatus::kOkay;
    while (!stop_rx_) {  // Can also do && net_xport_.get_status() ==
                         // NetCoreStatus::kOkay
      populated_bytes_ = rx_buffer_.size();
      rx_buffer_.resize(size_);

      ssize_t bytes_received = net_xport_.wait_and_receive(
          rx_buffer_.data() + populated_bytes_, size_ - populated_bytes_);

      if (bytes_received > 0) {
        if (static_cast<size_t>(bytes_received) < size_) {
          rx_buffer_.resize(static_cast<size_t>(bytes_received));
        }
        std::vector<S> rasters;
        size_t bytes_deserialized = S::deserialize(rx_buffer_, rasters);
        q.enqueue_bulk(rasters.begin(),
                       rasters.size());
        rx_buffer_.erase(rx_buffer_.begin(),
                         rx_buffer_.begin() + bytes_deserialized);

      } else if (bytes_received < 0) {
        if (++fail_count_ > kFailLimit) {
          status_ = ReceiverStatus::kError;
          rx_buffer_.clear();
          break;
        }
      }
    }

    return ReceiverStatus::kStopped;
  }
  1. epoll-based receive. I've broken the network stack class into 3 stages:
  • constructor: creates a socket, binds
  • wait_for_connection: listens, accepts, establishes sockets of interest for polling
  • wait_and_receive: sleeps and wakes up when there's data to populate in the rx buffer

Basically, am I "epolling" correctly here? The code works, but I'd like to hear from folks with more experience using it.

ssize_t LinuxTCPCore::wait_and_receive(unsigned char* buffer,
                                       size_t num_bytes) noexcept
{
  if (num_bytes == 0) {
    return 0;
  }
  int nfds = epoll_wait(epoll_fd_, events_, kMaxEvents, -1);
  if (nfds < 0) {
    status_ = NetCoreStatus::kError;
  }
  ssize_t bytes_received = 0;
  ssize_t curr_received = 0;
  for (size_t i = 0; i < static_cast<size_t>(nfds); ++i) {
    if (events_[i].data.fd == comm_socket_) {
      curr_received = read(comm_socket_, buffer, num_bytes);
      // EPOLLET requires that we continue reading from the FD until
      // EWOULDBLOCK/EAGAIN
      if (curr_received < 0) {
        if (errno == EWOULDBLOCK || errno == EAGAIN) {
          break;
        } else {
          status_ = NetCoreStatus::kError;
          return -1;
        }
      } else {
        buffer += curr_received;
        bytes_received += curr_received;
      }
    }
  }
  status_ = NetCoreStatus::kOkay;
  return bytes_received;
}
  1. deserialization of events times (vector<unsigned char> -> vector<uint64_t>).

I hate the memcpy and the C-style cast, but I found them to be much more legible than an attempt I made with reinterpret_cast. Originally, I was destroying the buffer and returning the vector of SpikeRasters, but I decided to allow the above generic receive routine to own and manage that memory instead.

I use a 4-byte delimiter word to separate the individual structs during serialization. Should I encode a size instead? Should I use both?

  static size_t deserialize(const std::vector<unsigned char>& buff,
                            std::vector<SpikeRaster<T>>& result)
  {
    auto delim_start = kDelimiter.begin();
    auto delim_end = kDelimiter.end();
    auto range_start = buff.begin();
    auto range_end = buff.end();
    auto found = range_end;
    size_t bytes_deserialized = 0;
    while ((found = std::search(range_start, range_end, delim_start,
                                delim_end)) != range_end) {
      // deserialize the id
      auto id = *(T*)buff.data();
      range_start += sizeof(id);

      // deserialize the raster event times
      auto raster_bytes = static_cast<size_t>(found - range_start);
      std::vector<T> current(raster_bytes / sizeof(T), 0);
      memcpy(current.data(), buff.data() + sizeof(id), raster_bytes);
      result.emplace_back(SpikeRaster<T>{id, std::move(current)});

      // housekeeping for the return value, and updating the start
      // iterator
      bytes_deserialized += (sizeof(id) + raster_bytes + kDelimiter.size());
      range_start = found + kDelimiter.size();
    }

    return bytes_deserialized;
  }

I've tried to provide a little bit of scoping for the review above, but any and all feedback on the repo is appreciated, especially in regards to class/template design, legibility, and idiomatic usage.

\$\endgroup\$

2 Answers 2

1
\$\begingroup\$

Receiving network data

I don't understand why wait_and_receive() has a for-loop. Do you expect data from multiple sockets to be received? If so, this function is broken in many ways: num_bytes is never decreased so multiple sockets being ready could cause a buffer overflow, and data from multiple sockets can get interleaved in unexpected ways.

If nfds is only ever supposed to be 1 (excluding error conditions), then there is no need to use epoll at all, just do a blocking read() directly on the socket.

Note that the size of TCP packets that are being transmitted over the network do not have to match the size of a write() call. Packets can be split and reassembled by the sending operating system, by routers on the Internet, and by the receiving operating system. Even if you send() 4 bytes, the other side might have to do four recv() calls that each return 1 byte. The only guarantee you have is that the data is in the same order. Your code might seem to handle that, but you are still missng some corner cases.

Consider that you recv() a packet that has the delimiter, followed by 7 bytes, just one byte short of an uint64_t. In deserialize(), found will be true, and raster_bytes will be 7. But you allocate a vector of zero elements (raster_bytes / sizeof(uint64_t) is 0). The memcpy() will overwrite memory that wasn't allocated, which might cause your program to crash, or worse: corrupt memory that will lead to problems later. When it returns, result.size() == 0, so the caller thinks no values were received, but it will erase() all the bytes received so far, so whatever values were between the received delimiter and the next one will all be lost.

There is also the issue of the arbitrarily chosen delimiter of 4 bytes. Those bytes might be part of a valid raster event time. This is a very fragile way of adding delimiters to your data, and could cause your program to misinterpret the data. What you should instead do is:

  • Add a fixed-size header before each batch of raster event times.
  • The header should contain the number of event times following it.
  • The receiver should first set the receive buffer size to the size of the header, and wait for it to be filled.
  • You then have a complete header, from which you can read the number of event times that will follow.
  • You can then resize the receive buffer so it can accomodate all the event times.
  • Wait for the buffer to fill again, then process the whole buffer in one go.

You can still add a delimiter in the header, this way you could detect corrupted transmissions, but that would be very rare (TCP already has a checksum for each packet that the operating system will generate and check for you).

If you accept multiple connections, use one buffer per connection to avoid data getting mixed up.

Performance

Calling resize() often on a std::vector is most likely fine; it won't actually shrink the allocated memory unless you call shrink_to_fit().

However, if you erase() something from the start, it has to copy the remaining data over to the start. If it's very likely you'll erase everything anyway, it is fine, but if it often happens you have a lot of data left, then all that copying can become a performance issue. Ideally, you never partially erase the vector. You can do what I mentioned above and resize it exactly to the expected amount of data, so after processing everything you can just clear() the vector. Alternatively, you could use a ringbuffer or something similar to avoid copying data unnecessarily.

But apart from that, there is lots of explicit copying in your code. Data from the rx_buffer_ is memcpy()'d into the vector rasters, after which it is immediately copied into the queue q. Ideally, after you've read the header and you know how many raster event times you are going to receive, you'd read() them directly into the queue.

Strict aliasing violation

Depending on what the type T actually is, the following code is a strict aliasing violation:

// deserialize the id
auto id = *(T*)buff.data();

Consider that buff.data() might not be correctly aligned for T. Depending on T and the exact CPU architecture your code is running on, this could either cause a bus error, or it could read the data incorrectly without raising any error. The correct way to handle this is to memcpy() the data:

// deserialize the id
T id;
memcpy(&id, buff.data(), sizeof id);

If you hate memcpy(), you can use std::copy_n() for a more C++ way of doing things.

\$\endgroup\$
2
  • \$\begingroup\$ G. Sliepen and @J_H Thank you very much for the feedback. This is more than I expected to receive and I'm grateful. Knee deep in some work at the moment but I will get back to you soon. \$\endgroup\$
    – Ders
    Sep 3, 2023 at 4:44
  • \$\begingroup\$ amazing. Thank you for the improved rx algorithm. I plan to tackle it soon. I expected an element count would probably be the best approach. Although TCP provides a checksum, I may consider using a checksum for other transport implementations instead of a delimiter. re: multiple sockets, that is an eventuality I'd like to consider, but probably best to simplify for now and cross that bridge later. \$\endgroup\$
    – Ders
    Sep 7, 2023 at 21:50
2
\$\begingroup\$

git usage

... critiques on git maintenance

I see a feature branch, lots of little commits with one-line comments, and a .pre-commit-config.yaml config, that's all great!

However, I notice a month-old feature branch still lying around. Once a feature is done, merge it to main and delete the old branch. (This, of course, still preserves the edit history.)

For the message "Commit some Python WIP and black formatting changes.", prefer the verb "Add", since everything in the log is a commit. And if a mechanical reformat, e.g. from black, produces a bunch of changes, consider making that a separate commit to keep the history cleaner. Similarly for "rename" refactors which touch a bunch of code.

The repo includes some python source which does not appear in the OP. Pep-8 asks for three import sections, and this really does help the Gentle Reader. Recommend that you $ isort . every now and again. It's available from pypi. I like to have these items in my .isort.cfg file:

include_trailing_comma:  true
multi_line_output:       VERTICAL_HANGING_INDENT

public identifiers

I understand the use of Q. But S and T seem too short for use at global scope. Consider renaming both of them.


Hungarian notation

This project apparently uses a "konstant" prefix for many identifiers. If there's a coding standard you expect maintenance engineers to adhere to, write it down, preferably with an URL citation.

In the course of doing that, you might choose to re-evaluate whether such practice has historically had observable benefits to the project. If not, consider dropping the policy and doing some renames.


repeated failures

It's unclear why we accept as many as kFailLimit errors before bailing. Please add some comments describing expected results and the high level business motivation for ignoring multiple errors.

You will need automated tests for this functionality, and a way for tests to deliberately inject some number of failures. Possibly receive() is doing too much and another method should handle the error policy. Using a fixed limit that is independent of deserialized data size seems slightly surprising, but perhaps if I understood how the process which generates errors works then it would make more sense.

I assume we never encounter bytes_received == 0, or at least won't see it repeatedly.

Consider reporting a failure by raising an exception.

Please add a one-sentence comment that describes the behavior of the receive() method.

a few posts I've read

I agree with you that the repeated resizing seems inconvenient. Please have a comment cite the URLs of one or more posts which motivate the current resizing approach.


code coverage

I don't believe you're running automated tests against all the code paths that thread through this source text.

  if (nfds < 0) {
    status_ = NetCoreStatus::kError;
  }
  ...
  for (...; i < ...(nfds); ++i) {
    ...
  }
  status_ = NetCoreStatus::kOkay;
  return bytes_received;

What?!?

Negative number of file descriptors is the error case, good. And we execute the loop body zero times.

But then we unconditionally overwrite Error with Okay?!? (Plus we correctly return that we received 0 bytes.)

Also, recommend you slightly reduce the scope of curr_received, by putting it within the loop block. No need to initialize it to zero, since we never examine that zero value.

I agree with all of G. Sliepen's review remarks, including the concern that probably this is over-complicated and the for loop could be dispensed with entirely.


document the contract

          return -1;

Please add a one-sentence comment to wait_and_receive() which spells out that it definitely does not promise to return number of bytes read, since that is overloaded with a negative value for the error case.

In my opinion this is a "hard to call" Public API, it is harder than it needs to be to correctly consume it. Consider reporting errors by raising an exception.

\$\endgroup\$
1
  • 1
    \$\begingroup\$ Thank you @J_H. I plan to implement your feedback. I certainly need more testing, clear comments, even better change control practices. \$\endgroup\$
    – Ders
    Sep 7, 2023 at 21:46

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.