TEST(JobSystem, JobSystemParallelChildren) {
    v = 0;

    JobSystem js;
    js.adopt();

    struct User {
        std::atomic_int calls = {0};
        void func(JobSystem&, JobSystem::Job*) {
            v++;
            calls++;
        };
    } j;

    JobSystem::Job* root = js.createJob<User, &User::func>(nullptr, &j);
    for (int i=0 ; i<256 ; i++) {
        JobSystem::Job* job = js.createJob<User, &User::func>(root, &j);
        js.run(job);
    }
    js.runAndWait(root);

    EXPECT_EQ(257, v.load());
    EXPECT_EQ(257, j.calls);

    js.emancipate();
}

    void testit(std::atomic_int& count)
    {
        std::default_random_engine generator;
        std::uniform_int_distribution<int> distribution(1, 10);
        auto sleep_time = std::bind(distribution, generator);

        std::this_thread::sleep_for(std::chrono::microseconds(sleep_time()));

        ++count;
        if(count.load() == 5) {
            g_condition.notify_one();
        }
    }

namespace glws {
static char TAG[]="apitrace";

static JavaVM *s_javaVM = nullptr;
static jobject s_activityObject = nullptr;
static std::mutex s_activityObjectMutex;

static jmethodID s_setSurfaceSizeMethodID = nullptr;

static jfieldID s_descriptor = nullptr;

typedef std::shared_ptr<ANativeWindow> AndroidWindow;
static AndroidWindow s_currentNativeWindow;
static std::atomic_int s_currentNativeWindowId(0);
static std::mutex s_currentNativeWindowMutex;

static int s_stdout_fd = -1;
static int s_stderr_fd = -1;
static std::mutex s_stateMutex;
static std::condition_variable s_stateWait;

static EGLDisplay eglDisplay = EGL_NO_DISPLAY;
static char const *eglExtensions = NULL;
static bool has_EGL_KHR_create_context = false;
struct ResourceTracker;

static std::mutex s_resourcesMutex;
static std::vector<ResourceTracker*> s_resources;

class FdWriter : public std::streambuf
{
public:
    FdWriter(int fd)
    {
        m_fd = fd;
        int opts = fcntl(m_fd, F_GETFL);
        opts = opts & (~O_NONBLOCK);
        fcntl(m_fd, F_SETFL, opts);
    }

    // basic_streambuf interface
protected:
    std::streamsize xsputn(const char_type *__s, std::streamsize __n)
    {
        std::streamsize ret = 0;
        while( __n ) {
            ssize_t written = write(m_fd, __s, __n);
            if (written > 0) {
                __n -= written;
                __s += written;
                ret += written;
            } else {
                switch (errno) {
                case EBADF:
                case EINVAL:
                case EPIPE:
                    std::exit(1);
                    break;
                }
            }
        }
        return ret;
    }

    int_type overflow(int_type __c)
    {
        return xsputn(reinterpret_cast<const char_type *>(&__c), 1) == 1 ? __c : traits_type::eof();
    }

private:
    int m_fd = -1;
};

static EGLenum
translateAPI(glprofile::Profile profile)
{
    switch (profile.api) {
    case glprofile::API_GL:
        return EGL_OPENGL_API;
    case glprofile::API_GLES:
        return EGL_OPENGL_ES_API;
    default:
        assert(0);
        return EGL_NONE;
    }
}


/* Must be called before
 *
 * - eglCreateContext
 * - eglGetCurrentContext
 * - eglGetCurrentDisplay
 * - eglGetCurrentSurface
 * - eglMakeCurrent (when its ctx parameter is EGL_NO_CONTEXT ),
 * - eglWaitClient
 * - eglWaitNative
 */
static void
bindAPI(EGLenum api)
//.........这里部分代码省略.........

int main()
{
    Barrier barrier1(3);
    Barrier barrier2(3);
    arrived_all = false;
    processed_all = true;
    arrived_count.store(0);
    processed_count.store(allowed_max);
    std::thread worker[50];
    for (int i = 0; i < 6; i++)
        worker[i] = std::thread(worker_thread, i, &barrier1, &barrier2);

    {
        std::lock_guard<std::mutex> lk(m);
        std::cout << "Back in main(), after initialing first thread batch " << data << std::endl;
    }


    // wait until worker dies finishes execution
    for (int i = 0; i < 6; i++)
        worker[i].join();

 //    for (int i = 5; i < 10; i++)
 //        worker[i] = std::thread(worker_thread, i);

 //    {
 //        std::lock_guard<std::mutex> lk(m);
    //     std::cout << "Back in main(), after initialing second thread batch." <<  std::endl;
    // }

 //    // wait until worker dies finishes execution
 //    for (int i = 5; i < 10; i++)
 //        worker[i].join();
    std::cout << "finished main(), data = " << data << std::endl;
}

void spring_time::sleep(bool forceThreadSleep)
{
	if (forceThreadSleep) {
		spring::this_thread::sleep_for(chrono::nanoseconds(toNanoSecsi()));
		return;
	}


	// for very short time intervals use a yielding loop (yield is ~5x more accurate than sleep(), check the UnitTest)
	if (toMicroSecsi() < (avgThreadSleepTimeMicroSecs + avgThreadYieldTimeMicroSecs * 5)) {
		const spring_time s = gettime();

		while ((gettime() - s) < *this)
			thread_yield();

		return;
	}

	// expected wakeup time
	const spring_time t0 = gettime() + *this;

	spring::this_thread::sleep_for(chrono::nanoseconds(toNanoSecsi()));

	const spring_time t1 = gettime();
	const spring_time dt = t1 - t0;

	if (t1 >= t0) {
		// yes, it's not 100% thread correct, but it's okay when 1 of 1 million writes is dropped
		int avg = avgThreadSleepTimeMicroSecs.load();
		int newAvg = mix<float>(avg, dt.toMicroSecsf(), 0.1f);
		avgThreadSleepTimeMicroSecs.store(newAvg);
	}
}

    inline bool update()
    {
        if (windowId.load() == s_currentNativeWindowId.load())
            return false;

        s_currentNativeWindowMutex.lock();
        window = s_currentNativeWindow;
        windowId.store(s_currentNativeWindowId.load());
        s_currentNativeWindowMutex.unlock();
        return true;
    }

void stepLeft() {
	std::unique_lock<std::mutex> lock(mutex);
	for (int i = 0; i < 10; ++i) {
		std::cout << "left" << std::endl;
		isWaiting.fetch_add(1);
		if (isWaiting.load() % 2 != 0) {
			condVar.notify_one();
		}
		else {
			condVar.wait(lock);
		}
	}
}

static void thread_yield()
{
	const spring_time t0 = spring_time::gettime();
	this_thread::yield();
	const spring_time t1 = spring_time::gettime();
	const spring_time dt = t1 - t0;

	if (t1 >= t0) {
		// yes, it's not 100% thread correct, but it's okay when 1 of 1 million writes is dropped
		int avg = avgThreadYieldTimeMicroSecs.load();
		int newAvg = mix<float>(avg, dt.toMicroSecsf(), 0.1f);
		avgThreadYieldTimeMicroSecs.store(newAvg);
	}
}

void HAL_CleanNotifier(HAL_NotifierHandle notifierHandle, int32_t* status) {
  {
    std::lock_guard<priority_recursive_mutex> sync(notifierMutex);
    auto notifier = notifierHandles.Get(notifierHandle);
    if (!notifier) return;

    // remove from list
    if (notifier->prev) notifier->prev->next = notifier->next;
    if (notifier->next) notifier->next->prev = notifier->prev;
    if (notifiers == notifier) notifiers = notifier->next;
    notifierHandles.Free(notifierHandle);

    if (notifier->threaded) {
      NotifierThreadOwner* owner =
          static_cast<NotifierThreadOwner*>(notifier->param);
      delete owner;
    }
  }

  if (notifierRefCount.fetch_sub(1) == 1) {
    std::lock_guard<priority_mutex> sync(notifierInterruptMutex);
    // if this was the last notifier, clean up alarm and manager
    if (notifierAlarm) {
      notifierAlarm->writeEnable(false, status);
      notifierAlarm = nullptr;
    }
    if (notifierManager) {
      notifierManager->disable(status);
      notifierManager = nullptr;
    }
    closestTrigger = UINT64_MAX;
  }
}

static void onSurfaceDestroyed(JNIEnv */*env*/, jobject /*object*/)
{
    s_currentNativeWindowMutex.lock();
    s_currentNativeWindow.reset();
    s_currentNativeWindowId.store(0);
    s_currentNativeWindowMutex.unlock();
}

Texture TextureBuilder::Build2DTexture(TextureBindpoint target, unsigned width, unsigned height, unsigned count,
                                       unsigned channels, TexturePixelFormat pixelFormat, bool useMipmaps)
{
    int current_id = id.fetch_add(1, std::memory_order_relaxed);
    Texture result;

    result.info.width = width;
    result.info.height = height;
    result.info.count = count;
    result.info.type = TextureType::Tex2D;
    result.info.target = target;
    result.info.targetPixelFormat = pixelFormat;
    result.info.channels = channels;
    result.info.name = std::string(NAME_PREFIX) + std::to_string(current_id);

    unsigned minSize = std::min(width, height);
    if (useMipmaps)
    {
        unsigned i = 0;
        while (minSize > 1)
        {
            minSize /= 2;
            i++;
        }
        result.info.mipmaps = i;
    }

    return result;
}

void* initializeNotifier(void (*process)(uint64_t, void*), void *param, int32_t *status)
{
	if (!process) {
		*status = NULL_PARAMETER;
		return nullptr;
	}
	if (!notifierAtexitRegistered.test_and_set())
		std::atexit(cleanupNotifierAtExit);
	if (notifierRefCount.fetch_add(1) == 0) {
		std::lock_guard<priority_mutex> sync(notifierInterruptMutex);
		// create manager and alarm if not already created
		if (!notifierManager) {
			notifierManager = new tInterruptManager(1 << kTimerInterruptNumber, false, status);
			notifierManager->registerHandler(alarmCallback, NULL, status);
			notifierManager->enable(status);
		}
		if (!notifierAlarm) notifierAlarm = tAlarm::create(status);
	}

	std::lock_guard<priority_recursive_mutex> sync(notifierMutex);
	// create notifier structure and add to list
	Notifier* notifier = new Notifier();
	notifier->prev = nullptr;
	notifier->next = notifiers;
	if (notifier->next) notifier->next->prev = notifier;
	notifier->param = param;
	notifier->process = process;
	notifiers = notifier;
	return notifier;
}

void cleanNotifier(void* notifier_pointer, int32_t *status)
{
	{
		std::lock_guard<priority_recursive_mutex> sync(notifierMutex);
		Notifier* notifier = (Notifier*)notifier_pointer;

		// remove from list and delete
		if (notifier->prev) notifier->prev->next = notifier->next;
		if (notifier->next) notifier->next->prev = notifier->prev;
		if (notifiers == notifier) notifiers = notifier->next;
		delete notifier;
	}

	if (notifierRefCount.fetch_sub(1) == 1) {
		std::lock_guard<priority_mutex> sync(notifierInterruptMutex);
		// if this was the last notifier, clean up alarm and manager
		if (notifierAlarm) {
			notifierAlarm->writeEnable(false, status);
			delete notifierAlarm;
			notifierAlarm = nullptr;
		}
		if (notifierManager) {
			notifierManager->disable(status);
			delete notifierManager;
			notifierManager = nullptr;
		}
		closestTrigger = UINT64_MAX;
	}
}

	void print(){
		if(size==0)
			printf("<empty>\n");
		for(int i = 0; i < size;i++)
			printf("%d ", arr[(head.load() + i)%capacity]);
		printf("\n");
	}

void worker_thread(int i, class Barrier& barrier1, class Barrier& barrier2)
{
    {
        // std::lock_guard<std::mutex> lk(m);
        std::lock_guard<std::mutex> lk_guard(m);
        std::cout << "Worker thread "  << i << " has arrived." << std::endl;
    }

    barrier1.wait();

    {
        std::lock_guard<std::mutex> lk_guard(m);
        std::cout << "counts changing2, processed_all: " << processed_all << ", count > max:" << (arrived_count < allowed_max) << std::endl;
        arrived_count++;
        processed_count--;
        std::cout << "counts changing3, processed: " << processed_count << ", arrived:" << arrived_count<< std::endl;
    }

    {
        std::lock_guard<std::mutex> lk_guard(m);
        if (arrived_count.load() >= allowed_max) {
            std::cout << "Enough arrived, about to open floodgate #1" << std::endl;
            arrived_all = true;
            processed_all = false;
            // for (int i = 0; i < allowed_max; i++) {
            //     cv1.notify_one();
            // }
        }
        std::cout << "about to reach 2nd barrier:" << arrived_count<< std::endl;

    }
    // std::unique_lock<std::mutex> lk1(barrier1);
    barrier2.wait();
    // cv1.wait(lk1, []{return (processed_count < allowed_max) && arrived_all;});
    {
        std::lock_guard<std::mutex> lk_guard(m);
        processed_count++;
        arrived_count--;
    }

    // critical section would go here... then increment processed_count

    // report after critical section
    {
        std::lock_guard<std::mutex> lk(m);
        std::cout << "Worker thread "  << i << " data processing completed" << std::endl;
        if (processed_count == allowed_max) {
            std::cout << "Group finished, about to open floodgate #2" << std::endl;
            processed_all = true;
            arrived_all = false;
            // for (int i = 0; i < allowed_max; i++) {
            //     cv2.notify_one();
            // }
        }
    }
}

Order::Order(int clientAssignedId, const std::string& account, const std::string& security,
		double price, int amount, Operation operation, OrderType type) :
	m_id(gs_id.fetch_add(1)),
	m_clientAssignedId(clientAssignedId),
	m_account(account),
	m_security(security),
	m_price(price),
	m_amount(amount),
	m_operation(operation),
	m_type(type),
	m_state(State::Unsubmitted)
{

}

	bool add(int val){
		enqLock.lock();
		if((size+1) > capacity){
			if(addremdbg) printf("af %d\n", val);
			enqLock.unlock();
			return false;
		}
		if(addremdbg) printf("ap %d\n", val);
		arr[tail] = val;
		tail = (tail.load()+1)%capacity;
		size++;
		enqLock.unlock();
		return true;
	}

inline void thread_pool::execute(delegate_type e)
{
  if (fc_.fetch_sub(1, ::std::memory_order_relaxed) <= 0)
  {
    spawn_thread();
  }
  // else do nothing

  {
    ::std::lock_guard<decltype(cm_)> l(cm_);

    delegates_.emplace_back(::std::move(e));
  }

  cv_.notify_one();
}

int main () {
  using namespace shmdata;

  {
    // direct access writer with one reader
    Writer w("/tmp/check-stress",
             sizeof(Frame),
             "application/x-check-shmdata",
             &logger);
    assert(w);
    // init
    {
      Frame frame;
      assert(w.copy_to_shm(&frame, sizeof(Frame)));
    }
    Reader r("/tmp/check-stress",
             [](void *data, size_t size){
               // auto frame = static_cast<Frame *>(data);
               // std::cout << "(0) new data for client "
               //           << frame->count
               //           << " (size " << size << ")"
               //           << std::endl;
             },
             nullptr,
             nullptr,
             &logger);
    std::cout << "one reader" << std::endl;
    assert(r);
    auto reader_handle = std::async(std::launch::async, reader);
    while (1 != done.load()) {
      //  the following is locking the shared memory for writing
      auto access = w.get_one_write_access();
      assert(access);
      access->notify_clients(sizeof(Frame));
      auto frame = static_cast<Frame *>(access->get_mem());
      frame->count++;
    }
    assert(reader_handle.get());
  }
  std::this_thread::sleep_for (std::chrono::milliseconds(1000));
  std::cout << "fin" << std::endl;
  return 0;
}

int main(int argc, char** argv)
{
    if (argc != 8)
    {
        printf("%s sAddr1 sAddr2 sAddr3 iChannelNum iWorkerNumPerChannel iCountPerWorker iEntityNum\n",
                argv[0]);
        exit(-1);
    }

    std::vector<std::string> vecIPList;
    for (int i = 1; i <= 3; ++i)
        vecIPList.push_back(argv[i]);

    int iChannelNum = atoi(argv[4]);
    int iWorkerNumPerChannel = atoi(argv[5]);
    int iCountPerWorker = atoi(argv[6]);
    uint64_t iEntityNum = atoi(argv[7]);

    clsUUIDGenerator::GetInstance()->Init();

    for (int i = 0; i < iChannelNum; ++i)
    {
        std::shared_ptr<clsClient> poClient(new clsClient(&vecIPList));
        for (int j = 0; j < iWorkerNumPerChannel; ++j)
        {
            clsRWStressTester* poTester = new clsRWStressTester(poClient, iCountPerWorker, iEntityNum);
            poTester->Start();
        }
    }

    while (s_iStopCnt.load() == 0)
    {
        sleep(1);
        s_oStat.Print();
    }

    printf("Fail %lu\n", s_iFailCnt.load());

    return 0;
}