pipe.hh

/*
 * This file is open source software, licensed to you under the terms
 * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
 * distributed with this work for additional information regarding copyright
 * ownership.  You may not use this file except in compliance with the License.
 *
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
/*
 * Copyright (C) 2015 Cloudius Systems, Ltd.
 */
 
#pragma once
 
#include <seastar/core/future.hh>
#include <seastar/core/queue.hh>
 
#include <seastar/util/std-compat.hh>
#include <seastar/util/modules.hh>
 
 
namespace seastar {
 
 
 
SEASTAR_MODULE_EXPORT
class broken_pipe_exception : public std::exception {
public:
    virtual const char* what() const noexcept {
        return "Broken pipe";
    }
};
 
SEASTAR_MODULE_EXPORT
class unread_overflow_exception : public std::exception {
public:
    virtual const char* what() const noexcept {
        return "pipe_reader::unread() overflow";
    }
};
 
namespace internal {
template <typename T>
class pipe_buffer {
private:
    queue<std::optional<T>> _buf;
    bool _read_open = true;
    bool _write_open = true;
public:
    pipe_buffer(size_t size) : _buf(size) {}
    future<std::optional<T>> read() {
        return _buf.pop_eventually();
    }
    future<> write(T&& data) {
        return _buf.push_eventually(std::move(data));
    }
    bool readable() const {
        return _write_open || !_buf.empty();
    }
    bool writeable() const {
        return _read_open;
    }
    bool close_read() {
        // If a writer blocking (on a full queue), need to stop it.
        if (_buf.full()) {
            _buf.abort(std::make_exception_ptr(broken_pipe_exception()));
        }
        _read_open = false;
        return !_write_open;
    }
    bool close_write() {
        // If the queue is empty, write the EOF (disengaged optional) to the
        // queue to wake a blocked reader. If the queue is not empty, there is
        // no need to write the EOF to the queue - the reader will return an
        // EOF when it sees that _write_open == false.
        if (_buf.empty()) {
            _buf.push({});
        }
        _write_open = false;
        return !_read_open;
    }
};
} // namespace internal
 
SEASTAR_MODULE_EXPORT_BEGIN
template <typename T>
class pipe;
 
template <typename T>
class pipe_reader {
private:
    internal::pipe_buffer<T> *_bufp;
    std::optional<T> _unread;
    pipe_reader(internal::pipe_buffer<T> *bufp) noexcept : _bufp(bufp) { }
    friend class pipe<T>;
public:
    future<std::optional<T>> read() {
        if (_unread) {
            auto ret = std::move(*_unread);
            _unread = {};
            return make_ready_future<std::optional<T>>(std::move(ret));
        }
        if (_bufp->readable()) {
            return _bufp->read();
        } else {
            return make_ready_future<std::optional<T>>();
        }
    }
    void unread(T&& item) {
        if (_unread) {
            throw unread_overflow_exception();
        }
        _unread = std::move(item);
    }
    ~pipe_reader() {
        if (_bufp && _bufp->close_read()) {
            delete _bufp;
        }
    }
    // Allow move, but not copy, of pipe_reader
    pipe_reader(pipe_reader&& other) noexcept : _bufp(other._bufp) {
        other._bufp = nullptr;
    }
    pipe_reader& operator=(pipe_reader&& other) noexcept {
        std::swap(_bufp, other._bufp);
        return *this;
    }
};
 
template <typename T>
class pipe_writer {
private:
    internal::pipe_buffer<T> *_bufp;
    pipe_writer(internal::pipe_buffer<T> *bufp) noexcept : _bufp(bufp) { }
    friend class pipe<T>;
public:
    future<> write(T&& data) {
        if (_bufp->writeable()) {
            return _bufp->write(std::move(data));
        } else {
            return make_exception_future<>(broken_pipe_exception());
        }
    }
    ~pipe_writer() {
        if (_bufp && _bufp->close_write()) {
            delete _bufp;
        }
    }
    // Allow move, but not copy, of pipe_writer
    pipe_writer(pipe_writer&& other) noexcept : _bufp(other._bufp) {
        other._bufp = nullptr;
    }
    pipe_writer& operator=(pipe_writer&& other) noexcept {
        std::swap(_bufp, other._bufp);
        return *this;
    }
};
 
template <typename T>
class pipe {
public:
    pipe_reader<T> reader;
    pipe_writer<T> writer;
    explicit pipe(size_t size) : pipe(new internal::pipe_buffer<T>(size)) { }
private:
    pipe(internal::pipe_buffer<T> *bufp) noexcept : reader(bufp), writer(bufp) { }
};
 
SEASTAR_MODULE_EXPORT_END
 
 
} // namespace seastar