/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */

/* vim:set ts=2 sw=2 sts=2 et cindent: */

/* This Source Code Form is subject to the terms of the Mozilla Public

 * License, v. 2.0. If a copy of the MPL was not distributed with this

 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

#include "nsSocketTransportService2.h"

#include "PollableEvent.h"

#include "mozilla/Assertions.h"

#include "mozilla/DebugOnly.h"

#include "mozilla/Logging.h"

#include "prerror.h"

#include "prio.h"

#include "private/pprio.h"

#include "prnetdb.h"

#ifdef XP_WIN

#include "ShutdownLayer.h"

#else

#include <fcntl.h>

#define USEPIPE 1

#endif

namespace mozilla {

namespace net {

#ifndef USEPIPE

static PRDescIdentity sPollableEventLayerIdentity;

static PRIOMethods    sPollableEventLayerMethods;

static PRIOMethods   *sPollableEventLayerMethodsPtr = nullptr;

static void LazyInitSocket()

{

  MOZ_ASSERT(OnSocketThread(), "not on socket thread");

  if (sPollableEventLayerMethodsPtr) {

    return;

  }

  sPollableEventLayerIdentity = PR_GetUniqueIdentity("PollableEvent Layer");

  sPollableEventLayerMethods = *PR_GetDefaultIOMethods();

  sPollableEventLayerMethodsPtr = &sPollableEventLayerMethods;

}

static bool NewTCPSocketPair(PRFileDesc *fd[], bool aSetRecvBuff)

{

  // this is a replacement for PR_NewTCPSocketPair that manually

  // sets the recv buffer to 64K. A windows bug (1248358)

  // can result in using an incompatible rwin and window

  // scale option on localhost pipes if not set before connect.

  SOCKET_LOG(("NewTCPSocketPair %s a recv buffer tuning\n", aSetRecvBuff ? "with" : "without"));

  PRFileDesc *listener = nullptr;

  PRFileDesc *writer = nullptr;

  PRFileDesc *reader = nullptr;

  PRSocketOptionData recvBufferOpt;

  recvBufferOpt.option = PR_SockOpt_RecvBufferSize;

  recvBufferOpt.value.recv_buffer_size = 65535;

  PRSocketOptionData nodelayOpt;

  nodelayOpt.option = PR_SockOpt_NoDelay;

  nodelayOpt.value.no_delay = true;

  PRSocketOptionData noblockOpt;

  noblockOpt.option = PR_SockOpt_Nonblocking;

  noblockOpt.value.non_blocking = true;

  listener = PR_OpenTCPSocket(PR_AF_INET);

  if (!listener) {

    goto failed;

  }

  if (aSetRecvBuff) {

    PR_SetSocketOption(listener, &recvBufferOpt);

  }

  PR_SetSocketOption(listener, &nodelayOpt);

  PRNetAddr listenAddr;

  memset(&listenAddr, 0, sizeof(listenAddr));

  if ((PR_InitializeNetAddr(PR_IpAddrLoopback, 0, &listenAddr) == PR_FAILURE) ||

      (PR_Bind(listener, &listenAddr) == PR_FAILURE) ||

      (PR_GetSockName(listener, &listenAddr) == PR_FAILURE) || // learn the dynamic port

      (PR_Listen(listener, 5) == PR_FAILURE)) {

    goto failed;

  }

  writer = PR_OpenTCPSocket(PR_AF_INET);

  if (!writer) {

    goto failed;

  }

  if (aSetRecvBuff) {

    PR_SetSocketOption(writer, &recvBufferOpt);

  }

  PR_SetSocketOption(writer, &nodelayOpt);

  PR_SetSocketOption(writer, &noblockOpt);

  PRNetAddr writerAddr;

  if (PR_InitializeNetAddr(PR_IpAddrLoopback, ntohs(listenAddr.inet.port), &writerAddr) == PR_FAILURE) {

    goto failed;

  }

  if (PR_Connect(writer, &writerAddr, PR_INTERVAL_NO_TIMEOUT) == PR_FAILURE) {

    if ((PR_GetError() != PR_IN_PROGRESS_ERROR) ||

        (PR_ConnectContinue(writer, PR_POLL_WRITE) == PR_FAILURE)) {

      goto failed;

    }

  }

  PR_SetFDInheritable(writer, false);

  reader = PR_Accept(listener, &listenAddr, PR_MillisecondsToInterval(200));

  if (!reader) {

    goto failed;

  }

  PR_SetFDInheritable(reader, false);

  if (aSetRecvBuff) {

    PR_SetSocketOption(reader, &recvBufferOpt);

  }

  PR_SetSocketOption(reader, &nodelayOpt);

  PR_SetSocketOption(reader, &noblockOpt);

  PR_Close(listener);

  fd[0] = reader;

  fd[1] = writer;

  return true;

failed:

  if (listener) {

    PR_Close(listener);

  }

  if (reader) {

    PR_Close(reader);

  }

  if (writer) {

    PR_Close(writer);

  }

  return false;

}

#endif

PollableEvent::PollableEvent()

  : mWriteFD(nullptr)

  , mReadFD(nullptr)

  , mSignaled(false)

  , mWriteFailed(false)

  , mSignalTimestampAdjusted(false)

{

  MOZ_COUNT_CTOR(PollableEvent);

  MOZ_ASSERT(OnSocketThread(), "not on socket thread");

  // create pair of prfiledesc that can be used as a poll()ble

  // signal. on windows use a localhost socket pair, and on

  // unix use a pipe.

#ifdef USEPIPE

  SOCKET_LOG(("PollableEvent() using pipe\n"));

  if (PR_CreatePipe(&mReadFD, &mWriteFD) == PR_SUCCESS) {

    // make the pipe non blocking. NSPR asserts at

    // trying to use SockOpt here

    PROsfd fd = PR_FileDesc2NativeHandle(mReadFD);

    int flags = fcntl(fd, F_GETFL, 0);

    (void)fcntl(fd, F_SETFL, flags | O_NONBLOCK);

    fd = PR_FileDesc2NativeHandle(mWriteFD);

    flags = fcntl(fd, F_GETFL, 0);

    (void)fcntl(fd, F_SETFL, flags | O_NONBLOCK);

  } else {

    mReadFD = nullptr;

    mWriteFD = nullptr;

    SOCKET_LOG(("PollableEvent() pipe failed\n"));

  }

#else

  SOCKET_LOG(("PollableEvent() using socket pair\n"));

  PRFileDesc *fd[2];

  LazyInitSocket();

  // Try with a increased recv buffer first (bug 1248358).

  if (NewTCPSocketPair(fd, true)) {

    mReadFD = fd[0];

    mWriteFD = fd[1];

  // If the previous fails try without recv buffer increase (bug 1305436).

  } else if (NewTCPSocketPair(fd, false)) {

    mReadFD = fd[0];

    mWriteFD = fd[1];

  // If both fail, try the old version.

  } else if (PR_NewTCPSocketPair(fd) == PR_SUCCESS) {

    mReadFD = fd[0];

    mWriteFD = fd[1];

    PRSocketOptionData socket_opt;

    DebugOnly<PRStatus> status;

    socket_opt.option = PR_SockOpt_NoDelay;

    socket_opt.value.no_delay = true;

    PR_SetSocketOption(mWriteFD, &socket_opt);

    PR_SetSocketOption(mReadFD, &socket_opt);

    socket_opt.option = PR_SockOpt_Nonblocking;

    socket_opt.value.non_blocking = true;

    status = PR_SetSocketOption(mWriteFD, &socket_opt);

    MOZ_ASSERT(status == PR_SUCCESS);

    status = PR_SetSocketOption(mReadFD, &socket_opt);

    MOZ_ASSERT(status == PR_SUCCESS);

  }

  if (mReadFD && mWriteFD) {

    // compatibility with LSPs such as McAfee that assume a NSPR

    // layer for read ala the nspr Pollable Event - Bug 698882. This layer is a nop.

    PRFileDesc *topLayer =

      PR_CreateIOLayerStub(sPollableEventLayerIdentity,

                           sPollableEventLayerMethodsPtr);

    if (topLayer) {

      if (PR_PushIOLayer(fd[0], PR_TOP_IO_LAYER, topLayer) == PR_FAILURE) {

        topLayer->dtor(topLayer);

      } else {

        SOCKET_LOG(("PollableEvent() nspr layer ok\n"));

        mReadFD = topLayer;

      }

    }

  } else {

    SOCKET_LOG(("PollableEvent() socketpair failed\n"));

  }

#endif

  if (mReadFD && mWriteFD) {

    // prime the system to deal with races invovled in [dc]tor cycle

    SOCKET_LOG(("PollableEvent() ctor ok\n"));

    mSignaled = true;

    MarkFirstSignalTimestamp();

    PR_Write(mWriteFD, "I", 1);

  }

}

PollableEvent::~PollableEvent()

{

  MOZ_COUNT_DTOR(PollableEvent);

  if (mWriteFD) {

#if defined(XP_WIN)

    AttachShutdownLayer(mWriteFD);

#endif

    PR_Close(mWriteFD);

  }

  if (mReadFD) {

#if defined(XP_WIN)

    AttachShutdownLayer(mReadFD);

#endif

    PR_Close(mReadFD);

  }

}

// we do not record signals on the socket thread

// because the socket thread can reliably look at its

// own runnable queue before selecting a poll time

// this is the "service the network without blocking" comment in

// nsSocketTransportService2.cpp

bool

PollableEvent::Signal()

{

  SOCKET_LOG(("PollableEvent::Signal\n"));

  if (!mWriteFD) {

    SOCKET_LOG(("PollableEvent::Signal Failed on no FD\n"));

    return false;

  }

#ifndef XP_WIN

  // On windows poll can hang and this became worse when we introduced the

  // patch for bug 698882 (see also bug 1292181), therefore we reverted the

  // behavior on windows to be as before bug 698882, e.g. write to the socket

  // also if an event dispatch is on the socket thread and writing to the

  // socket for each event. See bug 1292181.

  if (OnSocketThread()) {

    SOCKET_LOG(("PollableEvent::Signal OnSocketThread nop\n"));

    return true;

  }

#endif

#ifndef XP_WIN

  // To wake up the poll writing once is enough, but for Windows that can cause

  // hangs so we will write for every event.

  // For non-Windows systems it is enough to write just once.

  if (mSignaled) {

    return true;

  }

#endif

  if (!mSignaled) {

    mSignaled = true;

    MarkFirstSignalTimestamp();

  }

  int32_t status = PR_Write(mWriteFD, "M", 1);

  SOCKET_LOG(("PollableEvent::Signal PR_Write %d\n", status));

  if (status != 1) {

    NS_WARNING("PollableEvent::Signal Failed\n");

    SOCKET_LOG(("PollableEvent::Signal Failed\n"));

    mSignaled = false;

    mWriteFailed = true;

  } else {

    mWriteFailed = false;

  }

  return (status == 1);

}

bool

PollableEvent::Clear()

{

  // necessary because of the "dont signal on socket thread" optimization

  MOZ_ASSERT(OnSocketThread(), "not on socket thread");

  SOCKET_LOG(("PollableEvent::Clear\n"));

  if (!mFirstSignalAfterClear.IsNull()) {

    SOCKET_LOG(("PollableEvent::Clear time to signal %ums",

                (uint32_t)(TimeStamp::NowLoRes() - mFirstSignalAfterClear).ToMilliseconds()));

  }

  mFirstSignalAfterClear = TimeStamp();

  mSignalTimestampAdjusted = false;

  mSignaled = false;

  if (!mReadFD) {

    SOCKET_LOG(("PollableEvent::Clear mReadFD is null\n"));

    return false;

  }

  char buf[2048];

  int32_t status;

#ifdef XP_WIN

  // On Windows we are writing to the socket for each event, to be sure that we

  // do not have any deadlock read from the socket as much as we can.

  while (true) {

    status = PR_Read(mReadFD, buf, 2048);

    SOCKET_LOG(("PollableEvent::Clear PR_Read %d\n", status));

    if (status == 0) {

      SOCKET_LOG(("PollableEvent::Clear EOF!\n"));

      return false;

    }

    if (status < 0) {

      PRErrorCode code = PR_GetError();

      if (code == PR_WOULD_BLOCK_ERROR) {

        return true;

      } else {

        SOCKET_LOG(("PollableEvent::Clear unexpected error %d\n", code));

        return false;

      }

    }

  }

#else

  status = PR_Read(mReadFD, buf, 2048);

  SOCKET_LOG(("PollableEvent::Clear PR_Read %d\n", status));

  if (status == 1) {

    return true;

  }

  if (status == 0) {

    SOCKET_LOG(("PollableEvent::Clear EOF!\n"));

    return false;

  }

  if (status > 1) {

    MOZ_ASSERT(false);

    SOCKET_LOG(("PollableEvent::Clear Unexpected events\n"));

    Clear();

    return true;

  }

  PRErrorCode code = PR_GetError();

  if (code == PR_WOULD_BLOCK_ERROR) {

    return true;

  }

  SOCKET_LOG(("PollableEvent::Clear unexpected error %d\n", code));

  return false;

#endif //XP_WIN

}

void

PollableEvent::MarkFirstSignalTimestamp()

{

  if (mFirstSignalAfterClear.IsNull()) {

    SOCKET_LOG(("PollableEvent::MarkFirstSignalTimestamp"));

    mFirstSignalAfterClear = TimeStamp::NowLoRes();

  }

}

void

PollableEvent::AdjustFirstSignalTimestamp()

{

  if (!mSignalTimestampAdjusted && !mFirstSignalAfterClear.IsNull()) {

    SOCKET_LOG(("PollableEvent::AdjustFirstSignalTimestamp"));

    mFirstSignalAfterClear = TimeStamp::NowLoRes();

    mSignalTimestampAdjusted = true;

  }

}

bool

PollableEvent::IsSignallingAlive(TimeDuration const& timeout)

{

  if (mWriteFailed) {

    return false;

  }

#ifdef DEBUG

  // The timeout would be just a disturbance in a debug build.

  return true;

#else

  if (!mSignaled || mFirstSignalAfterClear.IsNull() || timeout == TimeDuration()) {

    return true;

  }

  TimeDuration delay = (TimeStamp::NowLoRes() - mFirstSignalAfterClear);

  bool timedOut = delay > timeout;

  return !timedOut;

#endif // DEBUG

}

} // namespace net

} // namespace mozilla