service.h

Go to the documentation of this file.

/*
 * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 * or in the "license" file accompanying this file. This file 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.
 */

#pragma once

#include <atomic>
#include <condition_variable>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <typeinfo>

#include <dataflow_lite/utils/observable_object.h>

enum ServiceState {
    CREATED,  // created and ready to start
    STARTED,  // started and ready to use
    SHUTDOWN,  // clean shutdown, possibly restartable
};

static std::map<ServiceState, std::string> SERVICE_STATE_NAME_MAP = {{CREATED, "CREATED"}, {STARTED, "STARTED"},
                                                                     {SHUTDOWN,"SHUTDOWN"}};

class Service {
public:

    Service() : state_(CREATED) {};
    virtual ~Service() = default;

    virtual bool start() {
      state_.setValue(STARTED);
      return true;
    }

    virtual bool shutdown() {
      state_.setValue(SHUTDOWN);
      return true;
    }

    virtual std::string getStatusString() {
      // a more descriptive name (tag supplied on construction) would be ideal
      return typeid(this).name() + std::string(", state=") + SERVICE_STATE_NAME_MAP[getState()];
    }

    ServiceState getState() {
      return state_.getValue();
    }

protected:
    void setState(ServiceState new_state) {
      state_.setValue(new_state);
    }

private:
    ObservableObject<ServiceState> state_;
};

//  consider extending the waiter interface
class RunnableService : public Service
{
public:
    RunnableService() {
      should_run_.store(false);
    }
    ~RunnableService() override = default;

    bool start() override {
      bool started = startWorkerThread();
      started &= Service::start();
      return started;
    }

    bool shutdown() override {
      bool is_shutdown = Service::shutdown();
      is_shutdown &= stopWorkerThread();
      return is_shutdown;
    }

    virtual bool isRunning() {
      return Service::getState() == ServiceState::STARTED && should_run_.load();
    }

    void waitForShutdown() {
      std::unique_lock <std::mutex> lck(this->mtx_);
      if (runnable_thread_.joinable()) {
        cv_.wait(lck); // todo guard against spurious wakeup, or could do while
      }
    }

    void waitForShutdown(std::chrono::milliseconds millis) {
      std::unique_lock <std::mutex> lck(this->mtx_);
      if (runnable_thread_.joinable()) {
        cv_.wait_for(lck, millis); // todo guard against spurious wakeup
      }
    }

    void join() {
      if (runnable_thread_.joinable()) {
        runnable_thread_.join();
      }
    }

    std::string getStatusString() override {
      return Service::getStatusString() + std::string(", isRunning=") + (isRunning()
        ? std::string("True") : std::string("False"));
    }

protected:

  virtual bool startWorkerThread() {
    if(!runnable_thread_.joinable()) {
      should_run_.store(true);
      runnable_thread_ = std::thread(&RunnableService::run, this);
      return true;
    }
    return false;
  }

  virtual bool stopWorkerThread() {
    if(should_run_.load()) {
      should_run_.store(false);
      return true;
    }
    return false;
  }

  virtual void run() {
    while(should_run_.load() && ServiceState::STARTED == Service::getState()) {
      work();
    }
    // done, notify anyone waiting
    std::unique_lock <std::mutex> lck(this->mtx_);
    this->cv_.notify_all();
  }

  virtual void work() = 0;

private:
  std::thread runnable_thread_;
  std::atomic<bool> should_run_{};
  std::condition_variable cv_;
  mutable std::mutex mtx_;
};