lib/cpp/src/thrift/windows/OverlappedSubmissionThread.cpp

/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); 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.
*/

#include <thrift/windows/OverlappedSubmissionThread.h>
#include <thrift/transport/TTransportException.h>
#include <boost/noncopyable.hpp>
#include <boost/scope_exit.hpp>
#include <process.h>

namespace apache {
namespace thrift {
namespace transport {

TOverlappedWorkItem::TOverlappedWorkItem()
  : SLIST_ENTRY(),
    action(UNKNOWN),
    h(INVALID_HANDLE_VALUE),
    buffer(NULL),
    buffer_len(0),
    overlap(),
    last_error(0),
    success(TRUE) {
}

void TOverlappedWorkItem::reset(uint8_t* buf, uint32_t len, HANDLE event) {
  memset(&overlap, 0, sizeof(overlap));
  overlap.hEvent = event;
  buffer = buf;
  buffer_len = len;
  last_error = 0;
  success = FALSE;
}

uint32_t TOverlappedWorkItem::overlappedResults(bool signal_failure) {
  DWORD bytes = 0;
  BOOL result = ::GetOverlappedResult(h, &overlap, &bytes, TRUE);
  if (signal_failure && !result) // get overlapped error case
  {
    GlobalOutput.perror("TPipe ::GetOverlappedResult errored GLE=", ::GetLastError());
    throw TTransportException(TTransportException::UNKNOWN, "TPipe: GetOverlappedResult failed");
  }
  return bytes;
}

bool TOverlappedWorkItem::process() {
  BOOST_SCOPE_EXIT((&doneSubmittingEvent)) { SetEvent(doneSubmittingEvent.h); }
  BOOST_SCOPE_EXIT_END

  switch (action) {
  case (CONNECT):
    success = ::ConnectNamedPipe(h, &overlap);
    if (success == FALSE)
      last_error = ::GetLastError();
    return true;
  case (READ):
    success = ::ReadFile(h, buffer, buffer_len, NULL, &overlap);
    if (success == FALSE)
      last_error = ::GetLastError();
    return true;
  case (CANCELIO):
    success = ::CancelIo(h);
    if (success == FALSE)
      last_error = ::GetLastError();
    return true;
  case (STOP):
  default:
    return false;
  }
}

void TOverlappedSubmissionThread::addWorkItem(TOverlappedWorkItem* item) {
  InterlockedPushEntrySList(&workList_, item);
  SetEvent(workAvailableEvent_.h);
  WaitForSingleObject(item->doneSubmittingEvent.h, INFINITE);
}

TOverlappedSubmissionThread* TOverlappedSubmissionThread::acquire_instance() {
  TAutoCrit lock(instanceGuard_);
  if (instance_ == NULL) {
    assert(instanceRefCount_ == 0);
    instance_ = new TOverlappedSubmissionThread;
  }
  ++instanceRefCount_;
  return instance_;
}
void TOverlappedSubmissionThread::release_instance() {
  TAutoCrit lock(instanceGuard_);
  if (--instanceRefCount_ == 0) {
    delete instance_;
    instance_ = NULL;
  }
}

TOverlappedSubmissionThread::TOverlappedSubmissionThread() {
  stopItem_.action = TOverlappedWorkItem::STOP;

  InitializeSListHead(&workList_);
  thread_ = (HANDLE)_beginthreadex(NULL, 0, thread_proc, this, 0, NULL);
  if (thread_ == 0) {
    GlobalOutput.perror("TOverlappedSubmissionThread unable to create thread, errno=", errno);
    throw TTransportException(TTransportException::NOT_OPEN,
                              " TOverlappedSubmissionThread unable to create thread");
  }
}

TOverlappedSubmissionThread::~TOverlappedSubmissionThread() {
  addWorkItem(&stopItem_);
  ::WaitForSingleObject(thread_, INFINITE);
  CloseHandle(thread_);
}

void TOverlappedSubmissionThread::run() {
  for (;;) {
    WaitForSingleObject(workAvailableEvent_.h, INFINITE);
    // todo check result
    SLIST_ENTRY* entry = NULL;
    while ((entry = InterlockedPopEntrySList(&workList_)) != NULL) {
      TOverlappedWorkItem& item = *static_cast<TOverlappedWorkItem*>(entry);
      if (!item.process())
        return;
    }
  }
}

unsigned __stdcall TOverlappedSubmissionThread::thread_proc(void* addr) {
  static_cast<TOverlappedSubmissionThread*>(addr)->run();
  return 0;
}

TCriticalSection TOverlappedSubmissionThread::instanceGuard_;
TOverlappedSubmissionThread* TOverlappedSubmissionThread::instance_;
uint32_t TOverlappedSubmissionThread::instanceRefCount_ = 0;
}
}
} // apach::thrift::transport