/*
 * Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

#ifndef SHARE_GC_Z_ZMESSAGEPORT_INLINE_HPP
#define SHARE_GC_Z_ZMESSAGEPORT_INLINE_HPP

#include "gc/z/zMessagePort.hpp"
#include "gc/z/zFuture.inline.hpp"
#include "gc/z/zList.inline.hpp"
#include "runtime/mutexLocker.hpp"

template <typename T>
class ZMessageRequest : public StackObj {
  friend class ZList<ZMessageRequest>;

private:
  T                          _message;
  uint64_t                   _seqnum;
  ZFuture<T>                 _result;
  ZListNode<ZMessageRequest> _node;

public:
  void initialize(T message, uint64_t seqnum) {
    _message = message;
    _seqnum = seqnum;
  }

  T message() const {
    return _message;
  }

  uint64_t seqnum() const {
    return _seqnum;
  }

  void wait() {
    const T message = _result.get();
    assert(message == _message, "Message mismatch");
  }

  void satisfy(T message) {
    _result.set(message);
  }
};

template <typename T>
inline ZMessagePort<T>::ZMessagePort() :
    _monitor(Monitor::leaf,
             "ZMessagePort",
             Monitor::_allow_vm_block_flag,
             Monitor::_safepoint_check_never),
    _has_message(false),
    _seqnum(0),
    _queue() {}

template <typename T>
inline void ZMessagePort<T>::send_sync(T message) {
  Request request;

  {
    // Enqueue message
    MonitorLocker ml(&_monitor, Monitor::_no_safepoint_check_flag);
    request.initialize(message, _seqnum);
    _queue.insert_last(&request);
    ml.notify();
  }

  // Wait for completion
  request.wait();

  {
    // Guard deletion of underlying semaphore. This is a workaround for a
    // bug in sem_post() in glibc < 2.21, where it's not safe to destroy
    // the semaphore immediately after returning from sem_wait(). The
    // reason is that sem_post() can touch the semaphore after a waiting
    // thread have returned from sem_wait(). To avoid this race we are
    // forcing the waiting thread to acquire/release the lock held by the
    // posting thread. https://sourceware.org/bugzilla/show_bug.cgi?id=12674
    MonitorLocker ml(&_monitor, Monitor::_no_safepoint_check_flag);
  }
}

template <typename T>
inline void ZMessagePort<T>::send_async(T message) {
  MonitorLocker ml(&_monitor, Monitor::_no_safepoint_check_flag);
  if (!_has_message) {
    // Post message
    _message = message;
    _has_message = true;
    ml.notify();
  }
}

template <typename T>
inline T ZMessagePort<T>::receive() {
  MonitorLocker ml(&_monitor, Monitor::_no_safepoint_check_flag);

  // Wait for message
  while (!_has_message && _queue.is_empty()) {
    ml.wait();
  }

  // Increment request sequence number
  _seqnum++;

  if (!_has_message) {
    // Message available in the queue
    _message = _queue.first()->message();
    _has_message = true;
  }

  return _message;
}

template <typename T>
inline void ZMessagePort<T>::ack() {
  MonitorLocker ml(&_monitor, Monitor::_no_safepoint_check_flag);

  if (!_has_message) {
    // Nothing to ack
    return;
  }

  // Satisfy requests (and duplicates) in queue
  ZListIterator<Request> iter(&_queue);
  for (Request* request; iter.next(&request);) {
    if (request->message() == _message && request->seqnum() < _seqnum) {
      // Dequeue and satisfy request. Note that the dequeue operation must
      // happen first, since the request will immediately be deallocated
      // once it has been satisfied.
      _queue.remove(request);
      request->satisfy(_message);
    }
  }

  if (_queue.is_empty()) {
    // Queue is empty
    _has_message = false;
  } else {
    // Post first message in queue
    _message = _queue.first()->message();
  }
}

inline void ZRendezvousPort::signal() {
  _port.send_sync(true /* ignored */);
}

inline void ZRendezvousPort::wait() {
  _port.receive();
}

inline void ZRendezvousPort::ack() {
  _port.ack();
}

#endif // SHARE_GC_Z_ZMESSAGEPORT_INLINE_HPP