netwerk/socket/nsNamedPipeIOLayer.cpp

   1 /* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* This Source Code Form is subject to the terms of the Mozilla Public
   3  * License, v. 2.0. If a copy of the MPL was not distributed with this
   4  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   5
   6 #include "nsNamedPipeIOLayer.h"
   7
   8 #include <algorithm>
   9 #include <utility>
  10
  11 #include "mozilla/Atomics.h"
  12 #include "mozilla/DebugOnly.h"
  13 #include "mozilla/Logging.h"
  14 #include "mozilla/RefPtr.h"
  15 #include "mozilla/Unused.h"
  16 #include "mozilla/net/DNS.h"
  17 #include "nsISupportsImpl.h"
  18 #include "nsNamedPipeService.h"
  19 #include "nsNativeCharsetUtils.h"
  20 #include "nsNetCID.h"
  21 #include "nsServiceManagerUtils.h"
  22 #include "nsSocketTransportService2.h"
  23 #include "nsString.h"
  24 #include "nsThreadUtils.h"
  25 #include "nspr.h"
  26 #include "private/pprio.h"
  27
  28 namespace mozilla {
  29 namespace net {
  30
  31 static mozilla::LazyLogModule gNamedPipeLog("NamedPipeWin");
  32 #define LOG_NPIO_DEBUG(...) \
  33   MOZ_LOG(gNamedPipeLog, mozilla::LogLevel::Debug, (__VA_ARGS__))
  34 #define LOG_NPIO_ERROR(...) \
  35   MOZ_LOG(gNamedPipeLog, mozilla::LogLevel::Error, (__VA_ARGS__))
  36
  37 PRDescIdentity nsNamedPipeLayerIdentity;
  38 static PRIOMethods nsNamedPipeLayerMethods;
  39
  40 class NamedPipeInfo final : public nsINamedPipeDataObserver {
  41  public:
  42   NS_DECL_THREADSAFE_ISUPPORTS
  43   NS_DECL_NSINAMEDPIPEDATAOBSERVER
  44
  45   explicit NamedPipeInfo();
  46
  47   nsresult Connect(const nsAString& aPath);
  48   nsresult Disconnect();
  49
  50   /**
  51    * Both blocking/non-blocking mode are supported in this class.
  52    * The default mode is non-blocking mode, however, the client may change its
  53    * mode to blocking mode during hand-shaking (e.g. nsSOCKSSocketInfo).
  54    *
  55    * In non-blocking mode, |Read| and |Write| should be called by clients only
  56    * when |GetPollFlags| reports data availability. That is, the client calls
  57    * |GetPollFlags| with |PR_POLL_READ| and/or |PR_POLL_WRITE| set, and
  58    * according to the flags that set, |GetPollFlags| will check buffers status
  59    * and decide corresponding actions:
  60    *
  61    * -------------------------------------------------------------------
  62    * |               | data in buffer          | empty buffer          |
  63    * |---------------+-------------------------+-----------------------|
  64    * | PR_POLL_READ  | out: PR_POLL_READ       | DoRead/DoReadContinue |
  65    * |---------------+-------------------------+-----------------------|
  66    * | PR_POLL_WRITE | DoWrite/DoWriteContinue | out: PR_POLL_WRITE    |
  67    * ------------------------------------------+------------------------
  68    *
  69    * |DoRead| and |DoWrite| initiate read/write operations asynchronously, and
  70    * the |DoReadContinue| and |DoWriteContinue| are used to check the amount
  71    * of the data are read/written to/from buffers.
  72    *
  73    * The output parameter and the return value of |GetPollFlags| are identical
  74    * because we don't rely on the low-level select function to wait for data
  75    * availability, we instead use nsNamedPipeService to poll I/O completeness.
  76    *
  77    * When client get |PR_POLL_READ| or |PR_POLL_WRITE| from |GetPollFlags|,
  78    * they are able to use |Read| or |Write| to access the data in the buffer,
  79    * and this is supposed to be very fast because no network traffic is
  80    * involved.
  81    *
  82    * In blocking mode, the flow is quite similar to non-blocking mode, but
  83    * |DoReadContinue| and |DoWriteContinue| are never been used since the
  84    * operations are done synchronously, which could lead to slow responses.
  85    */
  86   int32_t Read(void* aBuffer, int32_t aSize);
  87   int32_t Write(const void* aBuffer, int32_t aSize);
  88
  89   // Like Read, but doesn't remove data in internal buffer.
  90   uint32_t Peek(void* aBuffer, int32_t aSize);
  91
  92   // Number of bytes available to read in internal buffer.
  93   int32_t Available() const;
  94
  95   // Flush write buffer
  96   //
  97   // @return whether the buffer has been flushed
  98   bool Sync(uint32_t aTimeout);
  99   void SetNonblocking(bool nonblocking);
 100
 101   bool IsConnected() const;
 102   bool IsNonblocking() const;
 103   HANDLE GetHandle() const;
 104
 105   // Initiate and check current status for read/write operations.
 106   int16_t GetPollFlags(int16_t aInFlags, int16_t* aOutFlags);
 107
 108  private:
 109   virtual ~NamedPipeInfo();
 110
 111   /**
 112    * DoRead/DoWrite starts a read/write call synchronously or asynchronously
 113    * depending on |mNonblocking|. In blocking mode, they return when the action
 114    * has been done and in non-blocking mode it returns the number of bytes that
 115    * were read/written if the operation is done immediately. If it takes some
 116    * time to finish the operation, zero is returned and
 117    * DoReadContinue/DoWriteContinue must be called to get async I/O result.
 118    */
 119   int32_t DoRead();
 120   int32_t DoReadContinue();
 121   int32_t DoWrite();
 122   int32_t DoWriteContinue();
 123
 124   /**
 125    * There was a write size limitation of named pipe,
 126    * see https://support.microsoft.com/en-us/kb/119218 for more information.
 127    * The limitation no longer exists, so feel free to change the value.
 128    */
 129   static const uint32_t kBufferSize = 65536;
 130
 131   nsCOMPtr<nsINamedPipeService> mNamedPipeService;
 132
 133   HANDLE mPipe;                 // the handle to the named pipe.
 134   OVERLAPPED mReadOverlapped;   // used for asynchronous read operations.
 135   OVERLAPPED mWriteOverlapped;  // used for asynchronous write operations.
 136
 137   uint8_t mReadBuffer[kBufferSize];  // octets read from pipe.
 138
 139   /**
 140    * These indicates the [begin, end) position of the data in the buffer.
 141    */
 142   DWORD mReadBegin;
 143   DWORD mReadEnd;
 144
 145   bool mHasPendingRead;  // previous asynchronous read is not finished yet.
 146
 147   uint8_t mWriteBuffer[kBufferSize];  // octets to be written to pipe.
 148
 149   /**
 150    * These indicates the [begin, end) position of the data in the buffer.
 151    */
 152   DWORD mWriteBegin;  // how many bytes are already written.
 153   DWORD mWriteEnd;    // valid amount of data in the buffer.
 154
 155   bool mHasPendingWrite;  // previous asynchronous write is not finished yet.
 156
 157   /**
 158    * current blocking mode is non-blocking or not, accessed only in socket
 159    * thread.
 160    */
 161   bool mNonblocking;
 162
 163   Atomic<DWORD> mErrorCode;  // error code from Named Pipe Service.
 164 };
 165
 166 NS_IMPL_ISUPPORTS(NamedPipeInfo, nsINamedPipeDataObserver)
 167
 168 NamedPipeInfo::NamedPipeInfo()
 169     : mNamedPipeService(NamedPipeService::GetOrCreate()),
 170       mPipe(INVALID_HANDLE_VALUE),
 171       mReadBegin(0),
 172       mReadEnd(0),
 173       mHasPendingRead(false),
 174       mWriteBegin(0),
 175       mWriteEnd(0),
 176       mHasPendingWrite(false),
 177       mNonblocking(true),
 178       mErrorCode(0) {
 179   MOZ_ASSERT(mNamedPipeService);
 180
 181   ZeroMemory(&mReadOverlapped, sizeof(OVERLAPPED));
 182   ZeroMemory(&mWriteOverlapped, sizeof(OVERLAPPED));
 183 }
 184
 185 NamedPipeInfo::~NamedPipeInfo() { MOZ_ASSERT(!mPipe); }
 186
 187 // nsINamedPipeDataObserver
 188
 189 NS_IMETHODIMP
 190 NamedPipeInfo::OnDataAvailable(uint32_t aBytesTransferred, void* aOverlapped) {
 191   DebugOnly<bool> isOnPipeServiceThread;
 192   MOZ_ASSERT(NS_SUCCEEDED(mNamedPipeService->IsOnCurrentThread(
 193                  &isOnPipeServiceThread)) &&
 194              isOnPipeServiceThread);
 195
 196   if (aOverlapped == &mReadOverlapped) {
 197     LOG_NPIO_DEBUG("[%s] %p read %d bytes", __func__, this, aBytesTransferred);
 198   } else if (aOverlapped == &mWriteOverlapped) {
 199     LOG_NPIO_DEBUG("[%s] %p write %d bytes", __func__, this, aBytesTransferred);
 200   } else {
 201     MOZ_ASSERT(false, "invalid callback");
 202     mErrorCode = ERROR_INVALID_DATA;
 203     return NS_ERROR_FAILURE;
 204   }
 205
 206   mErrorCode = ERROR_SUCCESS;
 207
 208   // dispatch an empty event to trigger STS thread
 209   gSocketTransportService->Dispatch(
 210       NS_NewRunnableFunction("NamedPipeInfo::OnDataAvailable", [] {}),
 211       NS_DISPATCH_NORMAL);
 212
 213   return NS_OK;
 214 }
 215
 216 NS_IMETHODIMP
 217 NamedPipeInfo::OnError(uint32_t aError, void* aOverlapped) {
 218   DebugOnly<bool> isOnPipeServiceThread;
 219   MOZ_ASSERT(NS_SUCCEEDED(mNamedPipeService->IsOnCurrentThread(
 220                  &isOnPipeServiceThread)) &&
 221              isOnPipeServiceThread);
 222
 223   LOG_NPIO_ERROR("[%s] error code=%d", __func__, aError);
 224   mErrorCode = aError;
 225
 226   // dispatch an empty event to trigger STS thread
 227   gSocketTransportService->Dispatch(
 228       NS_NewRunnableFunction("NamedPipeInfo::OnError", [] {}),
 229       NS_DISPATCH_NORMAL);
 230
 231   return NS_OK;
 232 }
 233
 234 // Named pipe operations
 235
 236 nsresult NamedPipeInfo::Connect(const nsAString& aPath) {
 237   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 238
 239   HANDLE pipe =
 240       CreateFileW(PromiseFlatString(aPath).get(), GENERIC_READ | GENERIC_WRITE,
 241                   FILE_SHARE_READ | FILE_SHARE_WRITE, nullptr, OPEN_EXISTING,
 242                   FILE_FLAG_OVERLAPPED, nullptr);
 243
 244   if (pipe == INVALID_HANDLE_VALUE) {
 245     LOG_NPIO_ERROR("[%p] CreateFile error (%lu)", this, GetLastError());
 246     return NS_ERROR_FAILURE;
 247   }
 248
 249   DWORD pipeMode = PIPE_READMODE_MESSAGE;
 250   if (!SetNamedPipeHandleState(pipe, &pipeMode, nullptr, nullptr)) {
 251     LOG_NPIO_ERROR("[%p] SetNamedPipeHandleState error (%lu)", this,
 252                    GetLastError());
 253     CloseHandle(pipe);
 254     return NS_ERROR_FAILURE;
 255   }
 256
 257   nsresult rv = mNamedPipeService->AddDataObserver(pipe, this);
 258   if (NS_WARN_IF(NS_FAILED(rv))) {
 259     CloseHandle(pipe);
 260     return rv;
 261   }
 262
 263   HANDLE readEvent = CreateEventA(nullptr, TRUE, TRUE, "NamedPipeRead");
 264   if (NS_WARN_IF(!readEvent || readEvent == INVALID_HANDLE_VALUE)) {
 265     CloseHandle(pipe);
 266     return NS_ERROR_FAILURE;
 267   }
 268
 269   HANDLE writeEvent = CreateEventA(nullptr, TRUE, TRUE, "NamedPipeWrite");
 270   if (NS_WARN_IF(!writeEvent || writeEvent == INVALID_HANDLE_VALUE)) {
 271     CloseHandle(pipe);
 272     CloseHandle(readEvent);
 273     return NS_ERROR_FAILURE;
 274   }
 275
 276   mPipe = pipe;
 277   mReadOverlapped.hEvent = readEvent;
 278   mWriteOverlapped.hEvent = writeEvent;
 279   return NS_OK;
 280 }
 281
 282 nsresult NamedPipeInfo::Disconnect() {
 283   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 284
 285   nsresult rv = mNamedPipeService->RemoveDataObserver(mPipe, this);
 286   Unused << NS_WARN_IF(NS_FAILED(rv));
 287
 288   mPipe = nullptr;
 289
 290   if (mReadOverlapped.hEvent &&
 291       mReadOverlapped.hEvent != INVALID_HANDLE_VALUE) {
 292     CloseHandle(mReadOverlapped.hEvent);
 293     mReadOverlapped.hEvent = nullptr;
 294   }
 295
 296   if (mWriteOverlapped.hEvent &&
 297       mWriteOverlapped.hEvent != INVALID_HANDLE_VALUE) {
 298     CloseHandle(mWriteOverlapped.hEvent);
 299     mWriteOverlapped.hEvent = nullptr;
 300   }
 301
 302   return rv;
 303 }
 304
 305 int32_t NamedPipeInfo::Read(void* aBuffer, int32_t aSize) {
 306   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 307
 308   int32_t bytesRead = Peek(aBuffer, aSize);
 309
 310   if (bytesRead > 0) {
 311     mReadBegin += bytesRead;
 312   }
 313
 314   return bytesRead;
 315 }
 316
 317 int32_t NamedPipeInfo::Write(const void* aBuffer, int32_t aSize) {
 318   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 319   MOZ_ASSERT(mWriteBegin <= mWriteEnd);
 320
 321   if (!IsConnected()) {
 322     // pipe unconnected
 323     PR_SetError(PR_NOT_CONNECTED_ERROR, 0);
 324     return -1;
 325   }
 326
 327   if (mWriteBegin == mWriteEnd) {
 328     mWriteBegin = mWriteEnd = 0;
 329   }
 330
 331   int32_t bytesToWrite =
 332       std::min<int32_t>(aSize, sizeof(mWriteBuffer) - mWriteEnd);
 333   MOZ_ASSERT(bytesToWrite >= 0);
 334
 335   if (bytesToWrite == 0) {
 336     PR_SetError(IsNonblocking() ? PR_WOULD_BLOCK_ERROR : PR_IO_PENDING_ERROR,
 337                 0);
 338     return -1;
 339   }
 340
 341   memcpy(&mWriteBuffer[mWriteEnd], aBuffer, bytesToWrite);
 342   mWriteEnd += bytesToWrite;
 343
 344   /**
 345    * Triggers internal write operation by calling |GetPollFlags|.
 346    * This is required for callers that use blocking I/O because they don't call
 347    * |GetPollFlags| to write data, but this also works for non-blocking I/O.
 348    */
 349   int16_t outFlag;
 350   GetPollFlags(PR_POLL_WRITE, &outFlag);
 351
 352   return bytesToWrite;
 353 }
 354
 355 uint32_t NamedPipeInfo::Peek(void* aBuffer, int32_t aSize) {
 356   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 357   MOZ_ASSERT(mReadBegin <= mReadEnd);
 358
 359   if (!IsConnected()) {
 360     // pipe unconnected
 361     PR_SetError(PR_NOT_CONNECTED_ERROR, 0);
 362     return -1;
 363   }
 364
 365   /**
 366    * If there's nothing in the read buffer, try to trigger internal read
 367    * operation by calling |GetPollFlags|. This is required for callers that
 368    * use blocking I/O because they don't call |GetPollFlags| to read data,
 369    * but this also works for non-blocking I/O.
 370    */
 371   if (!Available()) {
 372     int16_t outFlag;
 373     GetPollFlags(PR_POLL_READ, &outFlag);
 374
 375     if (!(outFlag & PR_POLL_READ)) {
 376       PR_SetError(IsNonblocking() ? PR_WOULD_BLOCK_ERROR : PR_IO_PENDING_ERROR,
 377                   0);
 378       return -1;
 379     }
 380   }
 381
 382   // Available() can't return more than what fits to the buffer at the read
 383   // offset.
 384   int32_t bytesRead = std::min<int32_t>(aSize, Available());
 385   MOZ_ASSERT(bytesRead >= 0);
 386   MOZ_ASSERT(mReadBegin + bytesRead <= mReadEnd);
 387   memcpy(aBuffer, &mReadBuffer[mReadBegin], bytesRead);
 388   return bytesRead;
 389 }
 390
 391 int32_t NamedPipeInfo::Available() const {
 392   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 393   MOZ_ASSERT(mReadBegin <= mReadEnd);
 394   MOZ_ASSERT(mReadEnd - mReadBegin <= 0x7FFFFFFF);  // no more than int32_max
 395   return mReadEnd - mReadBegin;
 396 }
 397
 398 bool NamedPipeInfo::Sync(uint32_t aTimeout) {
 399   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 400   if (!mHasPendingWrite) {
 401     return true;
 402   }
 403   return WaitForSingleObject(mWriteOverlapped.hEvent, aTimeout) ==
 404          WAIT_OBJECT_0;
 405 }
 406
 407 void NamedPipeInfo::SetNonblocking(bool nonblocking) {
 408   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 409   mNonblocking = nonblocking;
 410 }
 411
 412 bool NamedPipeInfo::IsConnected() const {
 413   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 414   return mPipe && mPipe != INVALID_HANDLE_VALUE;
 415 }
 416
 417 bool NamedPipeInfo::IsNonblocking() const {
 418   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 419   return mNonblocking;
 420 }
 421
 422 HANDLE
 423 NamedPipeInfo::GetHandle() const {
 424   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 425   return mPipe;
 426 }
 427
 428 int16_t NamedPipeInfo::GetPollFlags(int16_t aInFlags, int16_t* aOutFlags) {
 429   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 430
 431   *aOutFlags = 0;
 432
 433   if (aInFlags & PR_POLL_READ) {
 434     int32_t bytesToRead = 0;
 435     if (mReadBegin < mReadEnd) {  // data in buffer and is ready to be read
 436       bytesToRead = Available();
 437     } else if (mHasPendingRead) {  // nonblocking I/O and has pending task
 438       bytesToRead = DoReadContinue();
 439     } else {  // read bufer is empty.
 440       bytesToRead = DoRead();
 441     }
 442
 443     if (bytesToRead > 0) {
 444       *aOutFlags |= PR_POLL_READ;
 445     } else if (bytesToRead < 0) {
 446       *aOutFlags |= PR_POLL_ERR;
 447     }
 448   }
 449
 450   if (aInFlags & PR_POLL_WRITE) {
 451     int32_t bytesWritten = 0;
 452     if (mHasPendingWrite) {  // nonblocking I/O and has pending task.
 453       bytesWritten = DoWriteContinue();
 454     } else if (mWriteBegin < mWriteEnd) {  // data in buffer, ready to write
 455       bytesWritten = DoWrite();
 456     } else {  // write buffer is empty.
 457       *aOutFlags |= PR_POLL_WRITE;
 458     }
 459
 460     if (bytesWritten < 0) {
 461       *aOutFlags |= PR_POLL_ERR;
 462     } else if (bytesWritten && !mHasPendingWrite && mWriteBegin == mWriteEnd) {
 463       *aOutFlags |= PR_POLL_WRITE;
 464     }
 465   }
 466
 467   return *aOutFlags;
 468 }
 469
 470 // @return: data has been read and is available
 471 int32_t NamedPipeInfo::DoRead() {
 472   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 473   MOZ_ASSERT(!mHasPendingRead);
 474   MOZ_ASSERT(mReadBegin == mReadEnd);  // the buffer should be empty
 475
 476   mReadBegin = 0;
 477   mReadEnd = 0;
 478
 479   BOOL success = ReadFile(mPipe, mReadBuffer, sizeof(mReadBuffer), &mReadEnd,
 480                           IsNonblocking() ? &mReadOverlapped : nullptr);
 481
 482   if (success) {
 483     LOG_NPIO_DEBUG("[%s][%p] %lu bytes read", __func__, this, mReadEnd);
 484     return mReadEnd;
 485   }
 486
 487   switch (GetLastError()) {
 488     case ERROR_MORE_DATA:  // has more data to read
 489       mHasPendingRead = true;
 490       return DoReadContinue();
 491
 492     case ERROR_IO_PENDING:  // read is pending
 493       mHasPendingRead = true;
 494       break;
 495
 496     default:
 497       LOG_NPIO_ERROR("[%s] ReadFile failed (%lu)", __func__, GetLastError());
 498       Disconnect();
 499       PR_SetError(PR_IO_ERROR, 0);
 500       return -1;
 501   }
 502
 503   return 0;
 504 }
 505
 506 int32_t NamedPipeInfo::DoReadContinue() {
 507   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 508   MOZ_ASSERT(mHasPendingRead);
 509   MOZ_ASSERT(mReadBegin == 0 && mReadEnd == 0);
 510
 511   BOOL success;
 512   success = GetOverlappedResult(mPipe, &mReadOverlapped, &mReadEnd, FALSE);
 513   if (success) {
 514     mHasPendingRead = false;
 515     if (mReadEnd == 0) {
 516       Disconnect();
 517       PR_SetError(PR_NOT_CONNECTED_ERROR, 0);
 518       return -1;
 519     }
 520
 521     LOG_NPIO_DEBUG("[%s][%p] %lu bytes read", __func__, this, mReadEnd);
 522     return mReadEnd;
 523   }
 524
 525   switch (GetLastError()) {
 526     case ERROR_MORE_DATA:
 527       mHasPendingRead = false;
 528       LOG_NPIO_DEBUG("[%s][%p] %lu bytes read", __func__, this, mReadEnd);
 529       return mReadEnd;
 530     case ERROR_IO_INCOMPLETE:  // still in progress
 531       break;
 532     default:
 533       LOG_NPIO_ERROR("[%s]: GetOverlappedResult failed (%lu)", __func__,
 534                      GetLastError());
 535       Disconnect();
 536       PR_SetError(PR_IO_ERROR, 0);
 537       return -1;
 538   }
 539
 540   return 0;
 541 }
 542
 543 int32_t NamedPipeInfo::DoWrite() {
 544   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 545   MOZ_ASSERT(!mHasPendingWrite);
 546   MOZ_ASSERT(mWriteBegin < mWriteEnd);
 547
 548   DWORD bytesWritten = 0;
 549   BOOL success =
 550       WriteFile(mPipe, &mWriteBuffer[mWriteBegin], mWriteEnd - mWriteBegin,
 551                 &bytesWritten, IsNonblocking() ? &mWriteOverlapped : nullptr);
 552
 553   if (success) {
 554     mWriteBegin += bytesWritten;
 555     LOG_NPIO_DEBUG("[%s][%p] %lu bytes written", __func__, this, bytesWritten);
 556     return bytesWritten;
 557   }
 558
 559   if (GetLastError() != ERROR_IO_PENDING) {
 560     LOG_NPIO_ERROR("[%s] WriteFile failed (%lu)", __func__, GetLastError());
 561     Disconnect();
 562     PR_SetError(PR_IO_ERROR, 0);
 563     return -1;
 564   }
 565
 566   mHasPendingWrite = true;
 567
 568   return 0;
 569 }
 570
 571 int32_t NamedPipeInfo::DoWriteContinue() {
 572   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 573   MOZ_ASSERT(mHasPendingWrite);
 574
 575   DWORD bytesWritten = 0;
 576   BOOL success =
 577       GetOverlappedResult(mPipe, &mWriteOverlapped, &bytesWritten, FALSE);
 578
 579   if (!success) {
 580     if (GetLastError() == ERROR_IO_INCOMPLETE) {
 581       // still in progress
 582       return 0;
 583     }
 584
 585     LOG_NPIO_ERROR("[%s] GetOverlappedResult failed (%lu)", __func__,
 586                    GetLastError());
 587     Disconnect();
 588     PR_SetError(PR_IO_ERROR, 0);
 589     return -1;
 590   }
 591
 592   mHasPendingWrite = false;
 593   mWriteBegin += bytesWritten;
 594   LOG_NPIO_DEBUG("[%s][%p] %lu bytes written", __func__, this, bytesWritten);
 595   return bytesWritten;
 596 }
 597
 598 static inline NamedPipeInfo* GetNamedPipeInfo(PRFileDesc* aFd) {
 599   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 600   MOZ_DIAGNOSTIC_ASSERT(aFd);
 601   MOZ_DIAGNOSTIC_ASSERT(aFd->secret);
 602   MOZ_DIAGNOSTIC_ASSERT(PR_GetLayersIdentity(aFd) == nsNamedPipeLayerIdentity);
 603
 604   if (!aFd || !aFd->secret ||
 605       PR_GetLayersIdentity(aFd) != nsNamedPipeLayerIdentity) {
 606     LOG_NPIO_ERROR("cannot get named pipe info");
 607     return nullptr;
 608   }
 609
 610   return reinterpret_cast<NamedPipeInfo*>(aFd->secret);
 611 }
 612
 613 static PRStatus nsNamedPipeConnect(PRFileDesc* aFd, const PRNetAddr* aAddr,
 614                                    PRIntervalTime aTimeout) {
 615   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 616
 617   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 618   if (!info) {
 619     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 620     return PR_FAILURE;
 621   }
 622
 623   nsAutoString path;
 624   if (NS_FAILED(NS_CopyNativeToUnicode(nsDependentCString(aAddr->local.path),
 625                                        path))) {
 626     PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
 627     return PR_FAILURE;
 628   }
 629   if (NS_WARN_IF(NS_FAILED(info->Connect(path)))) {
 630     return PR_FAILURE;
 631   }
 632
 633   return PR_SUCCESS;
 634 }
 635
 636 static PRStatus nsNamedPipeConnectContinue(PRFileDesc* aFd, PRInt16 aOutFlags) {
 637   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 638
 639   return PR_SUCCESS;
 640 }
 641
 642 static PRStatus nsNamedPipeClose(PRFileDesc* aFd) {
 643   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 644
 645   if (aFd->secret && PR_GetLayersIdentity(aFd) == nsNamedPipeLayerIdentity) {
 646     RefPtr<NamedPipeInfo> info = dont_AddRef(GetNamedPipeInfo(aFd));
 647     info->Disconnect();
 648     aFd->secret = nullptr;
 649     aFd->identity = PR_INVALID_IO_LAYER;
 650   }
 651
 652   MOZ_ASSERT(!aFd->lower);
 653   PR_Free(aFd);  // PRFileDescs are allocated with PR_Malloc().
 654
 655   return PR_SUCCESS;
 656 }
 657
 658 static PRInt32 nsNamedPipeSend(PRFileDesc* aFd, const void* aBuffer,
 659                                PRInt32 aAmount, PRIntn aFlags,
 660                                PRIntervalTime aTimeout) {
 661   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 662
 663   Unused << aFlags;
 664   Unused << aTimeout;
 665
 666   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 667   if (!info) {
 668     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 669     return -1;
 670   }
 671   return info->Write(aBuffer, aAmount);
 672 }
 673
 674 static PRInt32 nsNamedPipeRecv(PRFileDesc* aFd, void* aBuffer, PRInt32 aAmount,
 675                                PRIntn aFlags, PRIntervalTime aTimeout) {
 676   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 677
 678   Unused << aTimeout;
 679
 680   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 681   if (!info) {
 682     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 683     return -1;
 684   }
 685
 686   if (aFlags) {
 687     if (aFlags != PR_MSG_PEEK) {
 688       PR_SetError(PR_UNKNOWN_ERROR, 0);
 689       return -1;
 690     }
 691     return info->Peek(aBuffer, aAmount);
 692   }
 693
 694   return info->Read(aBuffer, aAmount);
 695 }
 696
 697 static inline PRInt32 nsNamedPipeRead(PRFileDesc* aFd, void* aBuffer,
 698                                       PRInt32 aAmount) {
 699   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 700
 701   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 702   if (!info) {
 703     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 704     return -1;
 705   }
 706   return info->Read(aBuffer, aAmount);
 707 }
 708
 709 static inline PRInt32 nsNamedPipeWrite(PRFileDesc* aFd, const void* aBuffer,
 710                                        PRInt32 aAmount) {
 711   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 712
 713   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 714   if (!info) {
 715     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 716     return -1;
 717   }
 718   return info->Write(aBuffer, aAmount);
 719 }
 720
 721 static PRInt32 nsNamedPipeAvailable(PRFileDesc* aFd) {
 722   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 723
 724   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 725   if (!info) {
 726     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 727     return -1;
 728   }
 729   return static_cast<PRInt32>(info->Available());
 730 }
 731
 732 static PRInt64 nsNamedPipeAvailable64(PRFileDesc* aFd) {
 733   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 734
 735   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 736   if (!info) {
 737     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 738     return -1;
 739   }
 740   return static_cast<PRInt64>(info->Available());
 741 }
 742
 743 static PRStatus nsNamedPipeSync(PRFileDesc* aFd) {
 744   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 745
 746   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 747   if (!info) {
 748     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 749     return PR_FAILURE;
 750   }
 751   return info->Sync(0) ? PR_SUCCESS : PR_FAILURE;
 752 }
 753
 754 static PRInt16 nsNamedPipePoll(PRFileDesc* aFd, PRInt16 aInFlags,
 755                                PRInt16* aOutFlags) {
 756   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 757
 758   NamedPipeInfo* info = GetNamedPipeInfo(aFd);
 759   if (!info) {
 760     PR_SetError(PR_BAD_DESCRIPTOR_ERROR, 0);
 761     return 0;
 762   }
 763   return info->GetPollFlags(aInFlags, aOutFlags);
 764 }
 765
 766 // FIXME: remove socket option functions?
 767 static PRStatus nsNamedPipeGetSocketOption(PRFileDesc* aFd,
 768                                            PRSocketOptionData* aData) {
 769   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 770
 771   MOZ_ASSERT(aFd);
 772   MOZ_ASSERT(aData);
 773
 774   switch (aData->option) {
 775     case PR_SockOpt_Nonblocking:
 776       aData->value.non_blocking =
 777           GetNamedPipeInfo(aFd)->IsNonblocking() ? PR_TRUE : PR_FALSE;
 778       break;
 779     case PR_SockOpt_Keepalive:
 780       aData->value.keep_alive = PR_TRUE;
 781       break;
 782     case PR_SockOpt_NoDelay:
 783       aData->value.no_delay = PR_TRUE;
 784       break;
 785     default:
 786       PR_SetError(PR_INVALID_METHOD_ERROR, 0);
 787       return PR_FAILURE;
 788   }
 789
 790   return PR_SUCCESS;
 791 }
 792
 793 static PRStatus nsNamedPipeSetSocketOption(PRFileDesc* aFd,
 794                                            const PRSocketOptionData* aData) {
 795   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 796
 797   MOZ_ASSERT(aFd);
 798   MOZ_ASSERT(aData);
 799
 800   switch (aData->option) {
 801     case PR_SockOpt_Nonblocking:
 802       GetNamedPipeInfo(aFd)->SetNonblocking(aData->value.non_blocking);
 803       break;
 804     case PR_SockOpt_Keepalive:
 805     case PR_SockOpt_NoDelay:
 806       break;
 807     default:
 808       PR_SetError(PR_INVALID_METHOD_ERROR, 0);
 809       return PR_FAILURE;
 810   }
 811
 812   return PR_SUCCESS;
 813 }
 814
 815 static void Initialize() {
 816   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 817
 818   static bool initialized = false;
 819   if (initialized) {
 820     return;
 821   }
 822
 823   nsNamedPipeLayerIdentity = PR_GetUniqueIdentity("Named Pipe layer");
 824   nsNamedPipeLayerMethods = *PR_GetDefaultIOMethods();
 825   nsNamedPipeLayerMethods.close = nsNamedPipeClose;
 826   nsNamedPipeLayerMethods.read = nsNamedPipeRead;
 827   nsNamedPipeLayerMethods.write = nsNamedPipeWrite;
 828   nsNamedPipeLayerMethods.available = nsNamedPipeAvailable;
 829   nsNamedPipeLayerMethods.available64 = nsNamedPipeAvailable64;
 830   nsNamedPipeLayerMethods.fsync = nsNamedPipeSync;
 831   nsNamedPipeLayerMethods.connect = nsNamedPipeConnect;
 832   nsNamedPipeLayerMethods.recv = nsNamedPipeRecv;
 833   nsNamedPipeLayerMethods.send = nsNamedPipeSend;
 834   nsNamedPipeLayerMethods.poll = nsNamedPipePoll;
 835   nsNamedPipeLayerMethods.getsocketoption = nsNamedPipeGetSocketOption;
 836   nsNamedPipeLayerMethods.setsocketoption = nsNamedPipeSetSocketOption;
 837   nsNamedPipeLayerMethods.connectcontinue = nsNamedPipeConnectContinue;
 838
 839   initialized = true;
 840 }
 841
 842 bool IsNamedPipePath(const nsACString& aPath) {
 843   return StringBeginsWith(aPath, "\\\\.\\pipe\\"_ns);
 844 }
 845
 846 PRFileDesc* CreateNamedPipeLayer() {
 847   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 848   Initialize();
 849
 850   PRFileDesc* layer =
 851       PR_CreateIOLayerStub(nsNamedPipeLayerIdentity, &nsNamedPipeLayerMethods);
 852   if (NS_WARN_IF(!layer)) {
 853     LOG_NPIO_ERROR("CreateNamedPipeLayer() failed.");
 854     return nullptr;
 855   }
 856
 857   RefPtr<NamedPipeInfo> info = new NamedPipeInfo();
 858   layer->secret = reinterpret_cast<PRFilePrivate*>(info.forget().take());
 859
 860   return layer;
 861 }
 862
 863 }  // namespace net
 864 }  // namespace mozilla