[ceph.git] / ceph / src / test / common / test_async_shared_mutex.cc

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2018 Red Hat
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation. See file COPYING.
 *
 */

#include "common/async/shared_mutex.h"
#include <optional>
#include <gtest/gtest.h>

namespace ceph::async {

using executor_type = boost::asio::io_context::executor_type;
using unique_lock = std::unique_lock<SharedMutex<executor_type>>;
using shared_lock = std::shared_lock<SharedMutex<executor_type>>;

// return a lambda that captures its error code and lock
auto capture(std::optional<boost::system::error_code>& ec, unique_lock& lock)
{
  return [&] (boost::system::error_code e, unique_lock l) {
    ec = e;
    lock = std::move(l);
  };
}
auto capture(std::optional<boost::system::error_code>& ec, shared_lock& lock)
{
  return [&] (boost::system::error_code e, shared_lock l) {
    ec = e;
    lock = std::move(l);
  };
}

TEST(SharedMutex, async_exclusive)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());

  std::optional<boost::system::error_code> ec1, ec2, ec3;
  unique_lock lock1, lock2, lock3;

  // request three exclusive locks
  mutex.async_lock(capture(ec1, lock1));
  mutex.async_lock(capture(ec2, lock2));
  mutex.async_lock(capture(ec3, lock3));

  EXPECT_FALSE(ec1); // no callbacks until poll()
  EXPECT_FALSE(ec2);
  EXPECT_FALSE(ec3);

  context.poll();
  EXPECT_FALSE(context.stopped()); // second lock still pending

  ASSERT_TRUE(ec1);
  EXPECT_EQ(boost::system::errc::success, *ec1);
  ASSERT_TRUE(lock1);
  EXPECT_FALSE(ec2);

  lock1.unlock();

  EXPECT_FALSE(ec2);

  context.poll();
  EXPECT_FALSE(context.stopped());

  ASSERT_TRUE(ec2);
  EXPECT_EQ(boost::system::errc::success, *ec2);
  ASSERT_TRUE(lock2);
  EXPECT_FALSE(ec3);

  lock2.unlock();

  EXPECT_FALSE(ec3);

  context.poll();
  EXPECT_TRUE(context.stopped());

  ASSERT_TRUE(ec3);
  EXPECT_EQ(boost::system::errc::success, *ec3);
  ASSERT_TRUE(lock3);
}

TEST(SharedMutex, async_shared)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());

  std::optional<boost::system::error_code> ec1, ec2;
  shared_lock lock1, lock2;

  // request two shared locks
  mutex.async_lock_shared(capture(ec1, lock1));
  mutex.async_lock_shared(capture(ec2, lock2));

  EXPECT_FALSE(ec1); // no callbacks until poll()
  EXPECT_FALSE(ec2);

  context.poll();
  EXPECT_TRUE(context.stopped());

  ASSERT_TRUE(ec1);
  EXPECT_EQ(boost::system::errc::success, *ec1);
  ASSERT_TRUE(lock1);
  ASSERT_TRUE(ec2);
  EXPECT_EQ(boost::system::errc::success, *ec2);
  ASSERT_TRUE(lock2);
}

TEST(SharedMutex, async_exclusive_while_shared)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());

  std::optional<boost::system::error_code> ec1, ec2;
  shared_lock lock1;
  unique_lock lock2;

  // request a shared and exclusive lock
  mutex.async_lock_shared(capture(ec1, lock1));
  mutex.async_lock(capture(ec2, lock2));

  EXPECT_FALSE(ec1); // no callbacks until poll()
  EXPECT_FALSE(ec2);

  context.poll();
  EXPECT_FALSE(context.stopped()); // second lock still pending

  ASSERT_TRUE(ec1);
  EXPECT_EQ(boost::system::errc::success, *ec1);
  ASSERT_TRUE(lock1);
  EXPECT_FALSE(ec2);

  lock1.unlock();

  EXPECT_FALSE(ec2);

  context.poll();
  EXPECT_TRUE(context.stopped());

  ASSERT_TRUE(ec2);
  EXPECT_EQ(boost::system::errc::success, *ec2);
  ASSERT_TRUE(lock2);
}

TEST(SharedMutex, async_shared_while_exclusive)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());

  std::optional<boost::system::error_code> ec1, ec2;
  unique_lock lock1;
  shared_lock lock2;

  // request an exclusive and shared lock
  mutex.async_lock(capture(ec1, lock1));
  mutex.async_lock_shared(capture(ec2, lock2));

  EXPECT_FALSE(ec1); // no callbacks until poll()
  EXPECT_FALSE(ec2);

  context.poll();
  EXPECT_FALSE(context.stopped()); // second lock still pending

  ASSERT_TRUE(ec1);
  EXPECT_EQ(boost::system::errc::success, *ec1);
  ASSERT_TRUE(lock1);
  EXPECT_FALSE(ec2);

  lock1.unlock();

  EXPECT_FALSE(ec2);

  context.poll();
  EXPECT_TRUE(context.stopped());

  ASSERT_TRUE(ec2);
  EXPECT_EQ(boost::system::errc::success, *ec2);
  ASSERT_TRUE(lock2);
}

TEST(SharedMutex, async_prioritize_exclusive)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());

  std::optional<boost::system::error_code> ec1, ec2, ec3;
  shared_lock lock1, lock3;
  unique_lock lock2;

  // acquire a shared lock, then request an exclusive and another shared lock
  mutex.async_lock_shared(capture(ec1, lock1));
  mutex.async_lock(capture(ec2, lock2));
  mutex.async_lock_shared(capture(ec3, lock3));

  EXPECT_FALSE(ec1); // no callbacks until poll()
  EXPECT_FALSE(ec2);
  EXPECT_FALSE(ec3);

  context.poll();
  EXPECT_FALSE(context.stopped());

  ASSERT_TRUE(ec1);
  EXPECT_EQ(boost::system::errc::success, *ec1);
  ASSERT_TRUE(lock1);
  EXPECT_FALSE(ec2);
  // exclusive waiter blocks the second shared lock
  EXPECT_FALSE(ec3);

  lock1.unlock();

  EXPECT_FALSE(ec2);
  EXPECT_FALSE(ec3);

  context.poll();
  EXPECT_FALSE(context.stopped());

  ASSERT_TRUE(ec2);
  EXPECT_EQ(boost::system::errc::success, *ec2);
  ASSERT_TRUE(lock2);
  EXPECT_FALSE(ec3);
}

TEST(SharedMutex, async_cancel)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());

  std::optional<boost::system::error_code> ec1, ec2, ec3, ec4;
  unique_lock lock1, lock2;
  shared_lock lock3, lock4;

  // request 2 exclusive and shared locks
  mutex.async_lock(capture(ec1, lock1));
  mutex.async_lock(capture(ec2, lock2));
  mutex.async_lock_shared(capture(ec3, lock3));
  mutex.async_lock_shared(capture(ec4, lock4));

  EXPECT_FALSE(ec1); // no callbacks until poll()
  EXPECT_FALSE(ec2);
  EXPECT_FALSE(ec3);
  EXPECT_FALSE(ec4);

  context.poll();
  EXPECT_FALSE(context.stopped());

  ASSERT_TRUE(ec1);
  EXPECT_EQ(boost::system::errc::success, *ec1);
  ASSERT_TRUE(lock1);
  EXPECT_FALSE(ec2);
  EXPECT_FALSE(ec3);
  EXPECT_FALSE(ec4);

  mutex.cancel();

  EXPECT_FALSE(ec2);
  EXPECT_FALSE(ec3);
  EXPECT_FALSE(ec4);

  context.poll();
  EXPECT_TRUE(context.stopped());

  ASSERT_TRUE(ec2);
  EXPECT_EQ(boost::asio::error::operation_aborted, *ec2);
  EXPECT_FALSE(lock2);
  ASSERT_TRUE(ec3);
  EXPECT_EQ(boost::asio::error::operation_aborted, *ec3);
  EXPECT_FALSE(lock3);
  ASSERT_TRUE(ec4);
  EXPECT_EQ(boost::asio::error::operation_aborted, *ec4);
  EXPECT_FALSE(lock4);
}

TEST(SharedMutex, async_destruct)
{
  boost::asio::io_context context;

  std::optional<boost::system::error_code> ec1, ec2, ec3, ec4;
  unique_lock lock1, lock2;
  shared_lock lock3, lock4;

  {
    SharedMutex mutex(context.get_executor());

    // request 2 exclusive and shared locks
    mutex.async_lock(capture(ec1, lock1));
    mutex.async_lock(capture(ec2, lock2));
    mutex.async_lock_shared(capture(ec3, lock3));
    mutex.async_lock_shared(capture(ec4, lock4));
  }

  EXPECT_FALSE(ec1); // no callbacks until poll()
  EXPECT_FALSE(ec2);
  EXPECT_FALSE(ec3);
  EXPECT_FALSE(ec4);

  context.poll();
  EXPECT_TRUE(context.stopped());

  ASSERT_TRUE(ec1);
  EXPECT_EQ(boost::system::errc::success, *ec1);
  ASSERT_TRUE(lock1);
  ASSERT_TRUE(ec2);
  EXPECT_EQ(boost::asio::error::operation_aborted, *ec2);
  EXPECT_FALSE(lock2);
  ASSERT_TRUE(ec3);
  EXPECT_EQ(boost::asio::error::operation_aborted, *ec3);
  EXPECT_FALSE(lock3);
  ASSERT_TRUE(ec4);
  EXPECT_EQ(boost::asio::error::operation_aborted, *ec4);
  EXPECT_FALSE(lock4);
}

// return a capture() lambda that's bound to the given executor
template <typename Executor, typename ...Args>
auto capture_ex(const Executor& ex, Args&& ...args)
{
  return boost::asio::bind_executor(ex, capture(std::forward<Args>(args)...));
}

TEST(SharedMutex, cross_executor)
{
  boost::asio::io_context mutex_context;
  SharedMutex mutex(mutex_context.get_executor());

  boost::asio::io_context callback_context;
  auto ex2 = callback_context.get_executor();

  std::optional<boost::system::error_code> ec1, ec2;
  unique_lock lock1, lock2;

  // request two exclusive locks
  mutex.async_lock(capture_ex(ex2, ec1, lock1));
  mutex.async_lock(capture_ex(ex2, ec2, lock2));

  EXPECT_FALSE(ec1);
  EXPECT_FALSE(ec2);

  mutex_context.poll();
  EXPECT_FALSE(mutex_context.stopped()); // maintains work on both executors

  EXPECT_FALSE(ec1); // no callbacks until poll() on callback_context
  EXPECT_FALSE(ec2);

  callback_context.poll();
  EXPECT_FALSE(callback_context.stopped()); // second lock still pending

  ASSERT_TRUE(ec1);
  EXPECT_EQ(boost::system::errc::success, *ec1);
  ASSERT_TRUE(lock1);
  EXPECT_FALSE(ec2);

  lock1.unlock();

  mutex_context.poll();
  EXPECT_TRUE(mutex_context.stopped());

  EXPECT_FALSE(ec2);

  callback_context.poll();
  EXPECT_TRUE(callback_context.stopped());

  ASSERT_TRUE(ec2);
  EXPECT_EQ(boost::system::errc::success, *ec2);
  ASSERT_TRUE(lock2);
}

TEST(SharedMutex, try_exclusive)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());
  {
    std::lock_guard lock{mutex};
    ASSERT_FALSE(mutex.try_lock()); // fail during exclusive
  }
  {
    std::shared_lock lock{mutex};
    ASSERT_FALSE(mutex.try_lock()); // fail during shared
  }
  ASSERT_TRUE(mutex.try_lock());
  mutex.unlock();
}

TEST(SharedMutex, try_shared)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());
  {
    std::lock_guard lock{mutex};
    ASSERT_FALSE(mutex.try_lock_shared()); // fail during exclusive
  }
  {
    std::shared_lock lock{mutex};
    ASSERT_TRUE(mutex.try_lock_shared()); // succeed during shared
    mutex.unlock_shared();
  }
  ASSERT_TRUE(mutex.try_lock_shared());
  mutex.unlock_shared();
}

TEST(SharedMutex, cancel)
{
  boost::asio::io_context context;
  SharedMutex mutex(context.get_executor());

  std::lock_guard l{mutex}; // exclusive lock blocks others

  // make synchronous lock calls in other threads
  auto f1 = std::async(std::launch::async, [&] { mutex.lock(); });
  auto f2 = std::async(std::launch::async, [&] { mutex.lock_shared(); });

  // this will race with spawned threads. just keep canceling until the
  // futures are ready
  const auto t = std::chrono::milliseconds(1);
  do { mutex.cancel(); } while (f1.wait_for(t) != std::future_status::ready);
  do { mutex.cancel(); } while (f2.wait_for(t) != std::future_status::ready);

  EXPECT_THROW(f1.get(), boost::system::system_error);
  EXPECT_THROW(f2.get(), boost::system::system_error);
}

} // namespace ceph::async
Commit	Line	Data
11fdf7f2 TL	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab
	3	/*
	4	* Ceph - scalable distributed file system
	5	*
	6	* Copyright (C) 2018 Red Hat
	7	*
	8	* This is free software; you can redistribute it and/or
	9	* modify it under the terms of the GNU Lesser General Public
	10	* License version 2.1, as published by the Free Software
	11	* Foundation. See file COPYING.
	12	*
	13	*/
	14
	15	#include "common/async/shared_mutex.h"
	16	#include <optional>
	17	#include <gtest/gtest.h>
	18
	19	namespace ceph::async {
	20
	21	using executor_type = boost::asio::io_context::executor_type;
	22	using unique_lock = std::unique_lock<SharedMutex<executor_type>>;
	23	using shared_lock = std::shared_lock<SharedMutex<executor_type>>;
	24
	25	// return a lambda that captures its error code and lock
	26	auto capture(std::optional<boost::system::error_code>& ec, unique_lock& lock)
	27	{
	28	return [&] (boost::system::error_code e, unique_lock l) {
	29	ec = e;
	30	lock = std::move(l);
	31	};
	32	}
	33	auto capture(std::optional<boost::system::error_code>& ec, shared_lock& lock)
	34	{
	35	return [&] (boost::system::error_code e, shared_lock l) {
	36	ec = e;
	37	lock = std::move(l);
	38	};
	39	}
	40
	41	TEST(SharedMutex, async_exclusive)
	42	{
	43	boost::asio::io_context context;
	44	SharedMutex mutex(context.get_executor());
	45
	46	std::optional<boost::system::error_code> ec1, ec2, ec3;
	47	unique_lock lock1, lock2, lock3;
	48
	49	// request three exclusive locks
	50	mutex.async_lock(capture(ec1, lock1));
	51	mutex.async_lock(capture(ec2, lock2));
	52	mutex.async_lock(capture(ec3, lock3));
	53
	54	EXPECT_FALSE(ec1); // no callbacks until poll()
	55	EXPECT_FALSE(ec2);
	56	EXPECT_FALSE(ec3);
	57
	58	context.poll();
	59	EXPECT_FALSE(context.stopped()); // second lock still pending
	60
	61	ASSERT_TRUE(ec1);
	62	EXPECT_EQ(boost::system::errc::success, *ec1);
	63	ASSERT_TRUE(lock1);
	64	EXPECT_FALSE(ec2);
65
66	lock1.unlock();
67
68	EXPECT_FALSE(ec2);
69
70	context.poll();
71	EXPECT_FALSE(context.stopped());
72
73	ASSERT_TRUE(ec2);
74	EXPECT_EQ(boost::system::errc::success, *ec2);
75	ASSERT_TRUE(lock2);
76	EXPECT_FALSE(ec3);
77
78	lock2.unlock();
79
80	EXPECT_FALSE(ec3);
81
82	context.poll();
83	EXPECT_TRUE(context.stopped());
84
85	ASSERT_TRUE(ec3);
86	EXPECT_EQ(boost::system::errc::success, *ec3);
87	ASSERT_TRUE(lock3);
88	}
89
90	TEST(SharedMutex, async_shared)
91	{
92	boost::asio::io_context context;
93	SharedMutex mutex(context.get_executor());
94
95	std::optional<boost::system::error_code> ec1, ec2;
96	shared_lock lock1, lock2;
97
98	// request two shared locks
99	mutex.async_lock_shared(capture(ec1, lock1));
100	mutex.async_lock_shared(capture(ec2, lock2));
101
102	EXPECT_FALSE(ec1); // no callbacks until poll()
103	EXPECT_FALSE(ec2);
104
105	context.poll();
106	EXPECT_TRUE(context.stopped());
107
108	ASSERT_TRUE(ec1);
109	EXPECT_EQ(boost::system::errc::success, *ec1);
110	ASSERT_TRUE(lock1);
111	ASSERT_TRUE(ec2);
112	EXPECT_EQ(boost::system::errc::success, *ec2);
113	ASSERT_TRUE(lock2);
114	}
115
116	TEST(SharedMutex, async_exclusive_while_shared)
117	{
118	boost::asio::io_context context;
119	SharedMutex mutex(context.get_executor());
120
121	std::optional<boost::system::error_code> ec1, ec2;
122	shared_lock lock1;
123	unique_lock lock2;
124
125	// request a shared and exclusive lock
126	mutex.async_lock_shared(capture(ec1, lock1));
127	mutex.async_lock(capture(ec2, lock2));
128
129	EXPECT_FALSE(ec1); // no callbacks until poll()
130	EXPECT_FALSE(ec2);
131
132	context.poll();
133	EXPECT_FALSE(context.stopped()); // second lock still pending
134
135	ASSERT_TRUE(ec1);
136	EXPECT_EQ(boost::system::errc::success, *ec1);
137	ASSERT_TRUE(lock1);
138	EXPECT_FALSE(ec2);
139
140	lock1.unlock();
141
142	EXPECT_FALSE(ec2);
143
144	context.poll();
145	EXPECT_TRUE(context.stopped());
146
147	ASSERT_TRUE(ec2);
148	EXPECT_EQ(boost::system::errc::success, *ec2);
149	ASSERT_TRUE(lock2);
150	}
151
152	TEST(SharedMutex, async_shared_while_exclusive)
153	{
154	boost::asio::io_context context;
155	SharedMutex mutex(context.get_executor());
156
157	std::optional<boost::system::error_code> ec1, ec2;
158	unique_lock lock1;
159	shared_lock lock2;
160
161	// request an exclusive and shared lock
162	mutex.async_lock(capture(ec1, lock1));
163	mutex.async_lock_shared(capture(ec2, lock2));
164
165	EXPECT_FALSE(ec1); // no callbacks until poll()
166	EXPECT_FALSE(ec2);
167
168	context.poll();
169	EXPECT_FALSE(context.stopped()); // second lock still pending
170
171	ASSERT_TRUE(ec1);
172	EXPECT_EQ(boost::system::errc::success, *ec1);
173	ASSERT_TRUE(lock1);
174	EXPECT_FALSE(ec2);
175
176	lock1.unlock();
177
178	EXPECT_FALSE(ec2);
179
180	context.poll();
181	EXPECT_TRUE(context.stopped());
182
183	ASSERT_TRUE(ec2);
184	EXPECT_EQ(boost::system::errc::success, *ec2);
185	ASSERT_TRUE(lock2);
186	}
187
188	TEST(SharedMutex, async_prioritize_exclusive)
189	{
190	boost::asio::io_context context;
191	SharedMutex mutex(context.get_executor());
192
193	std::optional<boost::system::error_code> ec1, ec2, ec3;
194	shared_lock lock1, lock3;
195	unique_lock lock2;
196
197	// acquire a shared lock, then request an exclusive and another shared lock
198	mutex.async_lock_shared(capture(ec1, lock1));
199	mutex.async_lock(capture(ec2, lock2));
200	mutex.async_lock_shared(capture(ec3, lock3));
201
202	EXPECT_FALSE(ec1); // no callbacks until poll()
203	EXPECT_FALSE(ec2);
204	EXPECT_FALSE(ec3);
205
206	context.poll();
207	EXPECT_FALSE(context.stopped());
208
209	ASSERT_TRUE(ec1);
210	EXPECT_EQ(boost::system::errc::success, *ec1);
211	ASSERT_TRUE(lock1);
212	EXPECT_FALSE(ec2);
213	// exclusive waiter blocks the second shared lock
214	EXPECT_FALSE(ec3);
215
216	lock1.unlock();
217
218	EXPECT_FALSE(ec2);
219	EXPECT_FALSE(ec3);
220
221	context.poll();
222	EXPECT_FALSE(context.stopped());
223
224	ASSERT_TRUE(ec2);
225	EXPECT_EQ(boost::system::errc::success, *ec2);
226	ASSERT_TRUE(lock2);
227	EXPECT_FALSE(ec3);
228	}
229
230	TEST(SharedMutex, async_cancel)
231	{
232	boost::asio::io_context context;
233	SharedMutex mutex(context.get_executor());
234
235	std::optional<boost::system::error_code> ec1, ec2, ec3, ec4;
236	unique_lock lock1, lock2;
237	shared_lock lock3, lock4;
238
239	// request 2 exclusive and shared locks
240	mutex.async_lock(capture(ec1, lock1));
241	mutex.async_lock(capture(ec2, lock2));
242	mutex.async_lock_shared(capture(ec3, lock3));
243	mutex.async_lock_shared(capture(ec4, lock4));
244
245	EXPECT_FALSE(ec1); // no callbacks until poll()
246	EXPECT_FALSE(ec2);
247	EXPECT_FALSE(ec3);
248	EXPECT_FALSE(ec4);
249
250	context.poll();
251	EXPECT_FALSE(context.stopped());
252
253	ASSERT_TRUE(ec1);
254	EXPECT_EQ(boost::system::errc::success, *ec1);
255	ASSERT_TRUE(lock1);
256	EXPECT_FALSE(ec2);
257	EXPECT_FALSE(ec3);
258	EXPECT_FALSE(ec4);
259
260	mutex.cancel();
261
262	EXPECT_FALSE(ec2);
263	EXPECT_FALSE(ec3);
264	EXPECT_FALSE(ec4);
265
266	context.poll();
267	EXPECT_TRUE(context.stopped());
268
269	ASSERT_TRUE(ec2);
270	EXPECT_EQ(boost::asio::error::operation_aborted, *ec2);
271	EXPECT_FALSE(lock2);
272	ASSERT_TRUE(ec3);
273	EXPECT_EQ(boost::asio::error::operation_aborted, *ec3);
274	EXPECT_FALSE(lock3);
275	ASSERT_TRUE(ec4);
276	EXPECT_EQ(boost::asio::error::operation_aborted, *ec4);
277	EXPECT_FALSE(lock4);
278	}
279
280	TEST(SharedMutex, async_destruct)
281	{
282	boost::asio::io_context context;
283
284	std::optional<boost::system::error_code> ec1, ec2, ec3, ec4;
285	unique_lock lock1, lock2;
286	shared_lock lock3, lock4;
287
288	{
289	SharedMutex mutex(context.get_executor());
290
291	// request 2 exclusive and shared locks
292	mutex.async_lock(capture(ec1, lock1));
293	mutex.async_lock(capture(ec2, lock2));
294	mutex.async_lock_shared(capture(ec3, lock3));
295	mutex.async_lock_shared(capture(ec4, lock4));
296	}
297
298	EXPECT_FALSE(ec1); // no callbacks until poll()
299	EXPECT_FALSE(ec2);
300	EXPECT_FALSE(ec3);
301	EXPECT_FALSE(ec4);
302
303	context.poll();
304	EXPECT_TRUE(context.stopped());
305
306	ASSERT_TRUE(ec1);
307	EXPECT_EQ(boost::system::errc::success, *ec1);
308	ASSERT_TRUE(lock1);
309	ASSERT_TRUE(ec2);
310	EXPECT_EQ(boost::asio::error::operation_aborted, *ec2);
311	EXPECT_FALSE(lock2);
312	ASSERT_TRUE(ec3);
313	EXPECT_EQ(boost::asio::error::operation_aborted, *ec3);
314	EXPECT_FALSE(lock3);
315	ASSERT_TRUE(ec4);
316	EXPECT_EQ(boost::asio::error::operation_aborted, *ec4);
317	EXPECT_FALSE(lock4);
318	}
319
320	// return a capture() lambda that's bound to the given executor
321	template <typename Executor, typename ...Args>
322	auto capture_ex(const Executor& ex, Args&& ...args)
323	{
324	return boost::asio::bind_executor(ex, capture(std::forward<Args>(args)...));
325	}
326
327	TEST(SharedMutex, cross_executor)
328	{
329	boost::asio::io_context mutex_context;
330	SharedMutex mutex(mutex_context.get_executor());
331
332	boost::asio::io_context callback_context;
333	auto ex2 = callback_context.get_executor();
334
335	std::optional<boost::system::error_code> ec1, ec2;
336	unique_lock lock1, lock2;
337
338	// request two exclusive locks
339	mutex.async_lock(capture_ex(ex2, ec1, lock1));
340	mutex.async_lock(capture_ex(ex2, ec2, lock2));
341
342	EXPECT_FALSE(ec1);
343	EXPECT_FALSE(ec2);
344
345	mutex_context.poll();
346	EXPECT_FALSE(mutex_context.stopped()); // maintains work on both executors
347
348	EXPECT_FALSE(ec1); // no callbacks until poll() on callback_context
349	EXPECT_FALSE(ec2);
350
351	callback_context.poll();
352	EXPECT_FALSE(callback_context.stopped()); // second lock still pending
353
354	ASSERT_TRUE(ec1);
355	EXPECT_EQ(boost::system::errc::success, *ec1);
356	ASSERT_TRUE(lock1);
357	EXPECT_FALSE(ec2);
358
359	lock1.unlock();
360
361	mutex_context.poll();
362	EXPECT_TRUE(mutex_context.stopped());
363
364	EXPECT_FALSE(ec2);
365
366	callback_context.poll();
367	EXPECT_TRUE(callback_context.stopped());
368
369	ASSERT_TRUE(ec2);
370	EXPECT_EQ(boost::system::errc::success, *ec2);
371	ASSERT_TRUE(lock2);
372	}
373
374	TEST(SharedMutex, try_exclusive)
375	{
376	boost::asio::io_context context;
377	SharedMutex mutex(context.get_executor());
378	{
379	std::lock_guard lock{mutex};
380	ASSERT_FALSE(mutex.try_lock()); // fail during exclusive
381	}
382	{
383	std::shared_lock lock{mutex};
384	ASSERT_FALSE(mutex.try_lock()); // fail during shared
385	}
386	ASSERT_TRUE(mutex.try_lock());
387	mutex.unlock();
388	}
389
390	TEST(SharedMutex, try_shared)
391	{
392	boost::asio::io_context context;
393	SharedMutex mutex(context.get_executor());
394	{
395	std::lock_guard lock{mutex};
396	ASSERT_FALSE(mutex.try_lock_shared()); // fail during exclusive
397	}
398	{
399	std::shared_lock lock{mutex};
400	ASSERT_TRUE(mutex.try_lock_shared()); // succeed during shared
401	mutex.unlock_shared();
402	}
403	ASSERT_TRUE(mutex.try_lock_shared());
404	mutex.unlock_shared();
405	}
406
407	TEST(SharedMutex, cancel)
408	{
409	boost::asio::io_context context;
410	SharedMutex mutex(context.get_executor());
411
412	std::lock_guard l{mutex}; // exclusive lock blocks others
413
414	// make synchronous lock calls in other threads
415	auto f1 = std::async(std::launch::async, [&] { mutex.lock(); });
416	auto f2 = std::async(std::launch::async, [&] { mutex.lock_shared(); });
417
418	// this will race with spawned threads. just keep canceling until the
419	// futures are ready
420	const auto t = std::chrono::milliseconds(1);
421	do { mutex.cancel(); } while (f1.wait_for(t) != std::future_status::ready);
422	do { mutex.cancel(); } while (f2.wait_for(t) != std::future_status::ready);
423
424	EXPECT_THROW(f1.get(), boost::system::system_error);
425	EXPECT_THROW(f2.get(), boost::system::system_error);
426	}
427
428	} // namespace ceph::async