Skip to content

Commit 9a050bc

Browse files
dante159753dante159753
committed
Revert "Fix Posix AIO timeout buffer lifetime"
This reverts commit 5f386ef.
1 parent 5f386ef commit 9a050bc

4 files changed

Lines changed: 13 additions & 121 deletions

File tree

‎ucm/store/posix/cc/aio_impl.cc‎

Lines changed: 11 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -139,12 +139,8 @@ static inline void AioSetEventFd(struct iocb* iocb, int32_t eventfd)
139139
iocb->aio_flags |= (1 << 0) /* IOCB_FLAG_RESFD */;
140140
iocb->aio_resfd = eventfd;
141141
}
142-
struct AioCompletionData {
143-
AioImpl::Callback callback;
144-
std::shared_ptr<void> bufferOwner;
145-
};
146142
static inline void ReleaseToAioCompletion(std::unique_ptr<struct iocb>& cb,
147-
std::unique_ptr<AioCompletionData>& data)
143+
std::unique_ptr<AioImpl::Callback>& data)
148144
{
149145
data.release();
150146
cb.release();
@@ -212,9 +208,7 @@ Status AioImpl::Setup(size_t timeoutMs)
212208
Status AioImpl::ReadAsync(Io&& io)
213209
{
214210
auto cb = std::make_unique<struct iocb>();
215-
auto data = std::make_unique<AioCompletionData>();
216-
data->callback = std::move(io.callback);
217-
data->bufferOwner = std::move(io.bufferOwner);
211+
auto data = std::make_unique<Callback>(std::move(io.callback));
218212
AioPrepareRead(cb.get(), io.fd, io.buffer, io.length, io.offset);
219213
cb->aio_data = (uintptr_t)(void*)data.get();
220214
Track(io.tag, cb.get());
@@ -231,9 +225,7 @@ Status AioImpl::ReadAsync(Io&& io)
231225
Status AioImpl::WriteAsync(Io&& io)
232226
{
233227
auto cb = std::make_unique<struct iocb>();
234-
auto data = std::make_unique<AioCompletionData>();
235-
data->callback = std::move(io.callback);
236-
data->bufferOwner = std::move(io.bufferOwner);
228+
auto data = std::make_unique<Callback>(std::move(io.callback));
237229
AioPrepareWrite(cb.get(), io.fd, io.buffer, io.length, io.offset);
238230
cb->aio_data = (uintptr_t)(void*)data.get();
239231
Track(io.tag, cb.get());
@@ -284,7 +276,7 @@ void AioImpl::HarvestCompletions(std::vector<io_event>& events)
284276
auto num = AioGetEvents(ctx_, 1, batchSize, events.data(), nullptr);
285277
for (auto i = 0; i < num; i++) {
286278
auto* iocbPtr = reinterpret_cast<struct iocb*>(events[i].obj);
287-
auto data = (AioCompletionData*)(void*)events[i].data;
279+
auto cb = (Callback*)(void*)events[i].data;
288280
Result res;
289281
if (events[i].res >= 0) {
290282
res.nBytes = events[i].res;
@@ -293,9 +285,9 @@ void AioImpl::HarvestCompletions(std::vector<io_event>& events)
293285
res.nBytes = -1;
294286
res.error = -static_cast<int>(events[i].res);
295287
}
296-
if (data) {
297-
data->callback(res);
298-
delete data;
288+
if (cb) {
289+
(*cb)(res);
290+
delete cb;
299291
}
300292
if (iocbPtr) {
301293
Untrack(iocbPtr);
@@ -355,10 +347,10 @@ void AioImpl::CancelTask(uint64_t tag)
355347
if (vec.empty()) { iocbTable_.erase(it); }
356348
}
357349
for (auto* cbp : cancelled) {
358-
auto* data = reinterpret_cast<AioCompletionData*>(cbp->aio_data);
359-
if (data) {
360-
data->callback(Result{-1, ECANCELED});
361-
delete data;
350+
auto* cb = reinterpret_cast<Callback*>(cbp->aio_data);
351+
if (cb) {
352+
(*cb)(Result{-1, ECANCELED});
353+
delete cb;
362354
}
363355
delete cbp;
364356
}

‎ucm/store/posix/cc/aio_impl.h‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,6 @@
2828
#include <cstdint>
2929
#include <functional>
3030
#include <linux/aio_abi.h>
31-
#include <memory>
3231
#include <mutex>
3332
#include <thread>
3433
#include <unordered_map>
@@ -59,7 +58,6 @@ class AioImpl {
5958
uint64_t offset;
6059
uint32_t length;
6160
void* buffer;
62-
std::shared_ptr<void> bufferOwner;
6361
Callback callback;
6462
uint64_t tag{0};
6563
};

‎ucm/store/posix/cc/io_engine_aio.h‎

Lines changed: 2 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -25,10 +25,7 @@
2525
#define UNIFIEDCACHE_POSIX_STORE_CC_IO_ENGINE_AIO_H
2626

2727
#include <cerrno>
28-
#include <cstdlib>
29-
#include <cstring>
3028
#include <limits>
31-
#include <memory>
3229
#include <mutex>
3330
#include <tuple>
3431
#include <unordered_map>
@@ -98,12 +95,6 @@ class IoEngineAio : public Detail::TaskWrapper<TransTask, Detail::TaskHandle> {
9895
static UC::Metrics::CachedMetric ioErrors{"posix_io_errors_total"};
9996
UC::Metrics::UpdateStats(ioErrors, 1.0);
10097
}
101-
std::shared_ptr<void> MakeOwnedIoBuffer() const
102-
{
103-
void* buffer = nullptr;
104-
if (posix_memalign(&buffer, 4096, shardSize_) != 0) { return {}; }
105-
return std::shared_ptr<void>(buffer, [](void* ptr) { std::free(ptr); });
106-
}
10798
void CommitBlock(Detail::BlockId id, bool success)
10899
{
109100
blockOperator_.Submit(BlockOperator::CommitTask{std::move(id), success});
@@ -151,33 +142,13 @@ class IoEngineAio : public Detail::TaskWrapper<TransTask, Detail::TaskHandle> {
151142
w->Done();
152143
return;
153144
}
154-
auto* userBuffer = shard.addrs.front();
155-
auto ioBuffer = userBuffer;
156-
std::shared_ptr<void> ioBufferOwner;
157-
if (timeoutMs_ > 0) {
158-
ioBufferOwner = MakeOwnedIoBuffer();
159-
if (!ioBufferOwner) {
160-
UC_ERROR("Failed to allocate owned aio buffer for task({}).", tid);
161-
handleFailure(result.fd);
162-
return;
163-
}
164-
ioBuffer = ioBufferOwner.get();
165-
if constexpr (dump) { std::memcpy(ioBuffer, userBuffer, shardSize_); }
166-
}
167145
AioImpl::Io io;
168146
io.fd = result.fd;
169147
io.offset = shard.index * shardSize_;
170148
io.length = shardSize_;
171-
io.buffer = ioBuffer;
172-
io.bufferOwner = ioBufferOwner;
149+
io.buffer = shard.addrs.front();
173150
io.tag = tid;
174-
io.callback = [this, tid, w, fd = result.fd, last, id, ioBuffer, userBuffer,
175-
copySize = shardSize_](AioImpl::Result ioResult) {
176-
if constexpr (!dump) {
177-
if (ioResult.error == 0 && !failureSet_.Contains(tid)) {
178-
std::memcpy(userBuffer, ioBuffer, copySize);
179-
}
180-
}
151+
io.callback = [this, tid, w, fd = result.fd, last, id](AioImpl::Result ioResult) {
181152
OnIoCallback<dump>(tid, w, fd, last, id, ioResult);
182153
};
183154
auto status = dump ? aio_.WriteAsync(std::move(io)) : aio_.ReadAsync(std::move(io));

‎ucm/store/test/case/posix/posix_store_test.cc‎

Lines changed: 0 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -316,75 +316,6 @@ TEST_F(UCPosixStoreTest, AioWaitTimesOutWhenCompletionIsLost)
316316
ASSERT_LT(std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count(), 1000);
317317
}
318318

319-
TEST_F(UCPosixStoreTest, AioDumpSubmitsOwnedBufferWhenTimeoutCanReturn)
320-
{
321-
using namespace UC::PosixStore;
322-
PosixStore store;
323-
ASSERT_EQ(store.Setup(MakeAioConfig(Path())), UC::Status::OK());
324-
ScopedAioHooks hooks;
325-
std::atomic<uintptr_t> submittedBuffer{0};
326-
TestHooks::SetAioSubmitHook([&submittedBuffer](aio_context_t, int64_t nr, iocb** ios) {
327-
submittedBuffer.store(ios[0]->aio_buf, std::memory_order_relaxed);
328-
return static_cast<int32_t>(nr);
329-
});
330-
TestHooks::SetAioCancelHook([](aio_context_t, struct iocb*, io_event*) { return 0; });
331-
auto buffer = MakeAlignedBuffer(8);
332-
ASSERT_NE(buffer.get(), nullptr);
333-
auto block = UC::Test::Detail::TypesHelper::MakeBlockIdRandomly();
334-
auto handle = store.Dump(MakeDumpDesc("AioDumpOwnedBuffer", block, buffer.get()));
335-
ASSERT_TRUE(handle.HasValue());
336-
337-
ASSERT_EQ(store.Wait(handle.Value()), UC::Status::Timeout());
338-
ASSERT_NE(submittedBuffer.load(std::memory_order_relaxed), 0U);
339-
ASSERT_NE(submittedBuffer.load(std::memory_order_relaxed),
340-
reinterpret_cast<uintptr_t>(buffer.get()));
341-
}
342-
343-
TEST_F(UCPosixStoreTest, AioLoadSubmitsOwnedBufferWhenTimeoutCanReturn)
344-
{
345-
using namespace UC::PosixStore;
346-
auto block = UC::Test::Detail::TypesHelper::MakeBlockIdRandomly();
347-
auto dumpBuffer = MakeAlignedBuffer(9);
348-
ASSERT_NE(dumpBuffer.get(), nullptr);
349-
{
350-
PosixStore store;
351-
ASSERT_EQ(store.Setup(MakeAioConfig(Path(), 5000)), UC::Status::OK());
352-
auto handle = store.Dump(MakeDumpDesc("AioLoadOwnedBufferPrepare", block, dumpBuffer.get()));
353-
ASSERT_TRUE(handle.HasValue());
354-
ASSERT_EQ(store.Wait(handle.Value()), UC::Status::OK());
355-
bool found = false;
356-
for (size_t i = 0; i < 20; ++i) {
357-
auto founds = store.Lookup(&block, 1);
358-
ASSERT_TRUE(founds.HasValue());
359-
if (founds.Value()[0]) {
360-
found = true;
361-
break;
362-
}
363-
std::this_thread::sleep_for(std::chrono::milliseconds(50));
364-
}
365-
ASSERT_TRUE(found);
366-
}
367-
368-
PosixStore store;
369-
ASSERT_EQ(store.Setup(MakeAioConfig(Path())), UC::Status::OK());
370-
ScopedAioHooks hooks;
371-
std::atomic<uintptr_t> submittedBuffer{0};
372-
TestHooks::SetAioSubmitHook([&submittedBuffer](aio_context_t, int64_t nr, iocb** ios) {
373-
submittedBuffer.store(ios[0]->aio_buf, std::memory_order_relaxed);
374-
return static_cast<int32_t>(nr);
375-
});
376-
TestHooks::SetAioCancelHook([](aio_context_t, struct iocb*, io_event*) { return 0; });
377-
auto loadBuffer = MakeAlignedBuffer(0);
378-
ASSERT_NE(loadBuffer.get(), nullptr);
379-
auto handle = store.Load(MakeDumpDesc("AioLoadOwnedBuffer", block, loadBuffer.get()));
380-
ASSERT_TRUE(handle.HasValue());
381-
382-
ASSERT_EQ(store.Wait(handle.Value()), UC::Status::Timeout());
383-
ASSERT_NE(submittedBuffer.load(std::memory_order_relaxed), 0U);
384-
ASSERT_NE(submittedBuffer.load(std::memory_order_relaxed),
385-
reinterpret_cast<uintptr_t>(loadBuffer.get()));
386-
}
387-
388319
TEST_F(UCPosixStoreTest, AioCheckFinishesLostCompletionAfterDeadline)
389320
{
390321
using namespace UC::PosixStore;

0 commit comments

Comments
 (0)