Inside Bitcoin

251         * @param[in] len Write up to this number of bytes.
252         * @param[in] flags Same as the flags of `recv(2)`. Just `MSG_PEEK` is honored.
253         * @return The number of bytes written to `buf`. `0` if `Eof()` has been called.
254         * If no bytes are available then `-1` is returned and `errno` is set to `EAGAIN`.
255         */
256        ssize_t GetBytes(void* buf, size_t len, int flags = 0) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
257
258        /**
259         * Deserialize a `CNetMessage` and remove it from the pipe.
260         * If not enough bytes are available then the function will wait. If parsing fails
261         * or EOF is signaled to the pipe, then `std::nullopt` is returned.
262         */
263        std::optional<CNetMessage> GetNetMsg() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
264
265        /**
266         * Push bytes to the pipe.
267         */
268        void PushBytes(const void* buf, size_t len) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
269
270        /**
271         * Construct and push CNetMessage to the pipe.
272         */
273        template <typename... Args>
274        void PushNetMsg(const std::string& type, Args&&... payload) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
275
276        /**
277         * Signal end-of-file on the receiving end (`GetBytes()` or `GetNetMsg()`).
278         */
279        void Eof() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
280
281    private:
282        /**
283         * Return when there is some data to read or EOF has been signaled.
284         * @param[in,out] lock Unique lock that must have been derived from `m_mutex` by `WAIT_LOCK(m_mutex, lock)`.
285         */
286        void WaitForDataOrEof(UniqueLock<Mutex>& lock) EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
287
288        Mutex m_mutex;
289        std::condition_variable m_cond;
290        std::vector<uint8_t> m_data GUARDED_BY(m_mutex);
291        bool m_eof GUARDED_BY(m_mutex){false};
292    };
293
294    struct Pipes {
295        Pipe recv;
296        Pipe send;
297    };
298
299    /**
300     * A basic thread-safe queue, used for queuing sockets to be returned by Accept().
301     */
302    class Queue
303    {
304    public:
305        using S = std::unique_ptr<DynSock>;
306
307        void Push(S s) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
308        {
309            LOCK(m_mutex);
310            m_queue.push(std::move(s));
311        }
312
313        std::optional<S> Pop() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
314        {
315            LOCK(m_mutex);
316            if (m_queue.empty()) {
317                return std::nullopt;
318            }
319            S front{std::move(m_queue.front())};
320            m_queue.pop();
321            return front;
322        }
323
324        bool Empty() const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
325        {
326            LOCK(m_mutex);
327            return m_queue.empty();
328        }
329
330    private:
331        mutable Mutex m_mutex;
332        std::queue<S> m_queue GUARDED_BY(m_mutex);
333    };
334
335    /**
336     * Create a new mocked sock.
337     * @param[in] pipes Send/recv pipes used by the Send() and Recv() methods.
338     * @param[in] accept_sockets Sockets to return by the Accept() method.
339     */
340    explicit DynSock(std::shared_ptr<Pipes> pipes, Queue* accept_sockets LIFETIMEBOUND);
341
342    /**
343     * Create a new mocked sock that represents a connected socket. It has pipes
344     * for data transport but there is no queue because connected sockets do
345     * not introduce new connected sockets.
346     * @param[in] pipes Send/recv pipes used by the Send() and Recv() methods.
347     */
348    explicit DynSock(std::shared_ptr<Pipes> pipes);
349
350    ~DynSock();
351
352    ssize_t Recv(void* buf, size_t len, int flags) const override;
353
354    ssize_t Send(const void* buf, size_t len, int) const override;
355
356    std::unique_ptr<Sock> Accept(sockaddr* addr, socklen_t* addr_len) const override;
357
358    bool Wait(std::chrono::milliseconds timeout,
359              Event requested,
360              Event* occurred = nullptr) const override;
361
362    bool WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock) const override;
363
364private:
365    DynSock& operator=(Sock&&) override;
366
367    std::shared_ptr<Pipes> m_pipes;
368    Queue* const m_accept_sockets;
369};
370
371template <typename... Args>
372void DynSock::Pipe::PushNetMsg(const std::string& type, Args&&... payload)
373{
374    auto msg = NetMsg::Make(type, std::forward<Args>(payload)...);
375    V1Transport transport{NodeId{0}};
376
377    const bool queued{transport.SetMessageToSend(msg)};
378    assert(queued);
379
380    LOCK(m_mutex);
381
382    for (;;) {
383        const auto& [bytes, _more, _msg_type] = transport.GetBytesToSend(/*have_next_message=*/true);
384        if (bytes.empty()) {
385            break;
386        }
387        m_data.insert(m_data.end(), bytes.begin(), bytes.end());
388        transport.MarkBytesSent(bytes.size());
389    }
390
391    m_cond.notify_all();
392}
393
394std::vector<NodeEvictionCandidate> GetRandomNodeEvictionCandidates(int n_candidates, FastRandomContext& random_context);
395
396#endif // BITCOIN_TEST_UTIL_NET_H
397