Skip to content

Commit 93a2050

Browse files
microcaimicrocai
committed
add an example unbufcpy3 based on unbufcpy1, but use coroutine!
1 parent a3ba5a8 commit 93a2050

File tree

3 files changed

+301
-1
lines changed

3 files changed

+301
-1
lines changed

‎example/unbufcpy/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11

22
add_executable(unbufcp1 unbufcp1/unbufcp1.c)
33
add_executable(unbufcp2 unbufcp2/unbufcp2.cpp)
4+
add_executable(unbufcp3 unbufcp3/unbufcp3.cpp)
45
set_target_properties(unbufcp1 PROPERTIES FOLDER "examples")
56
set_target_properties(unbufcp2 PROPERTIES FOLDER "examples")
7+
set_target_properties(unbufcp3 PROPERTIES FOLDER "examples")

‎example/unbufcpy/unbufcp3/unbufcp3.cpp‎

Lines changed: 294 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,294 @@
1+
/*++
2+
THIS CODE AND INFORMATION IS PROVIDED "AS IS" WITHOUT WARRANTY OF
3+
ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED
4+
TO THE IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A
5+
PARTICULAR PURPOSE.
6+
7+
Copyright (C) 2024 - 2025. microcai. All rights reserved.
8+
9+
Module Name:
10+
11+
unbufcp3.cpp
12+
13+
Abstract:
14+
15+
This single-threaded version shows how to multiplex I/O to a number of files with
16+
a single thread. This is the most efficient mechanism if you do not need the
17+
asynchronous completion model that the dual-threaded version offers.
18+
19+
Only one thread and one I/O completion port is used. The file handles for the
20+
source and destination file are both associated with the same port. The
21+
thread starts off by posting a number of overlapped
22+
reads from the source file. It then waits on the I/O completion port.
23+
Whenever a read completes, it immediately turns it around into a write to
24+
the destination file. Whenever a write completes, it immediately posts the
25+
next read from the source file.
26+
27+
Thread 1
28+
|
29+
|
30+
kick off a few
31+
overlapped reads
32+
|
33+
|
34+
->GetQueuedCompletionStatus(WritePort) <-----------
35+
| | |
36+
| |------------------------------- |
37+
| | | |
38+
| write has completed, read has completed, |
39+
| kick off another kick off the write. |
40+
| read | |
41+
| | | |
42+
|____| |_____________|
43+
44+
--*/
45+
#define BUFFER_SIZE (1024*1024*16)
46+
47+
#include "universal_async.hpp"
48+
#include "universal_fiber.hpp"
49+
50+
#include <chrono>
51+
52+
//
53+
// Define the size of the buffers used to do the I/O.
54+
// 64K is a nice number.
55+
//
56+
57+
//
58+
// The system's page size will always be a multiple of the
59+
// sector size. Do all I/Os in page-size chunks.
60+
//
61+
DWORD PageSize;
62+
63+
static void copy_coroutine(HANDLE IoPort, std::string sourcefilename, std::string destfilename)
64+
{
65+
ULARGE_INTEGER FileSize;
66+
ULARGE_INTEGER InitialSize;
67+
68+
auto StartTime = std::chrono::steady_clock::now();
69+
70+
//
71+
// Open the source file and create the destination file.
72+
// Use FILE_FLAG_NO_BUFFERING to avoid polluting the
73+
// system cache with two copies of the same data.
74+
//
75+
76+
HANDLE SourceFile = CreateFileA(sourcefilename.c_str(),
77+
GENERIC_READ | GENERIC_WRITE,
78+
FILE_SHARE_READ,
79+
NULL,
80+
OPEN_EXISTING,
81+
FILE_FLAG_NO_BUFFERING | FILE_FLAG_OVERLAPPED,
82+
NULL);
83+
if (SourceFile == INVALID_HANDLE_VALUE) {
84+
fprintf(stderr, "failed to open %s, error %d\n", sourcefilename.c_str(), GetLastError());
85+
exit(1);
86+
}
87+
FileSize.LowPart = GetFileSize(SourceFile, &FileSize.HighPart);
88+
if ((FileSize.LowPart == 0xffffffff) && (GetLastError() != NO_ERROR)) {
89+
fprintf(stderr, "GetFileSize failed, error %d\n", GetLastError());
90+
exit(1);
91+
}
92+
93+
HANDLE DestFile = CreateFileA(destfilename.c_str(),
94+
GENERIC_WRITE,
95+
FILE_SHARE_READ,
96+
NULL,
97+
CREATE_ALWAYS,
98+
FILE_FLAG_NO_BUFFERING | FILE_FLAG_OVERLAPPED,
99+
SourceFile);
100+
if (DestFile == INVALID_HANDLE_VALUE) {
101+
fprintf(stderr, "failed to open %s, error %d\n", destfilename.c_str(), GetLastError());
102+
exit(1);
103+
}
104+
105+
//
106+
// Extend the destination file so that the filesystem does not
107+
// turn our asynchronous writes into synchronous ones.
108+
//
109+
InitialSize.QuadPart = (FileSize.QuadPart + PageSize - 1) & ~((DWORD_PTR)(PageSize - 1));
110+
auto Status = SetFilePointer(DestFile,
111+
InitialSize.LowPart,
112+
(PLONG)&InitialSize.HighPart,
113+
FILE_BEGIN);
114+
if ((Status == INVALID_SET_FILE_POINTER) && (GetLastError() != NO_ERROR)) {
115+
fprintf(stderr, "initial SetFilePointer failed, error %d\n", GetLastError());
116+
exit(1);
117+
}
118+
auto Success = SetEndOfFile(DestFile);
119+
if (!Success) {
120+
fprintf(stderr, "SetEndOfFile failed, error %d\n", GetLastError());
121+
exit(1);
122+
}
123+
124+
SetFilePointer(DestFile, 0, 0, FILE_BEGIN);
125+
126+
bind_stackfull_iocp(SourceFile, IoPort);
127+
bind_stackfull_iocp(DestFile, IoPort);
128+
129+
FiberOVERLAPPED read_ov, write_ov;
130+
read_ov.set_offset(0);
131+
write_ov.set_offset(0);
132+
133+
char* pri_buf, *back_buf;
134+
std::array<char*, 2> buffer;
135+
136+
pri_buf = buffer[0] = (char*) malloc(BUFFER_SIZE);
137+
back_buf = buffer[1] = (char*) malloc(BUFFER_SIZE);
138+
139+
DWORD readLength = 0, back_readLength = 0, written = 0;
140+
auto ret = ReadFile(SourceFile, pri_buf, BUFFER_SIZE, &readLength, &read_ov);
141+
read_ov.last_error = GetLastError();
142+
if (!(!ret && read_ov.last_error != ERROR_IO_PENDING && read_ov.last_error != ERROR_MORE_DATA))
143+
readLength = get_overlapped_result(read_ov);
144+
145+
while (readLength > 0)
146+
{
147+
read_ov.add_offset(readLength);
148+
ret = ReadFile(SourceFile, back_buf, BUFFER_SIZE, &back_readLength, &read_ov);
149+
read_ov.last_error = GetLastError();
150+
151+
bool read_file_pending = !(!ret && read_ov.last_error != ERROR_IO_PENDING && read_ov.last_error != ERROR_MORE_DATA);
152+
153+
auto result = WriteFile(DestFile, pri_buf, (readLength + PageSize - 1) & ~(PageSize - 1), &written, &write_ov);
154+
write_ov.last_error = WSAGetLastError();
155+
if (!(!result && write_ov.last_error != ERROR_IO_PENDING))
156+
{
157+
written = get_overlapped_result(write_ov);
158+
}
159+
write_ov.add_offset(written);
160+
161+
if (write_ov.last_error)
162+
{
163+
if (read_file_pending)
164+
{
165+
// socket 发送错误，取消已经投递的文件读取请求
166+
CancelIoEx(SourceFile, &read_ov);
167+
// 无视取消是否成功，都等待文件读取请求。
168+
// 如果取消失败，（比如实际上文件已经读取成功）
169+
// 那么就当是无所谓了。
170+
// 如果取消成功，则 get_overlapped_result 会返回个错误
171+
// 但是不管 get_overlapped_result 返回的是啥，都已经无关紧要了
172+
// 这里还要调用 get_overlapped_result 仅仅是为了避免 退出本
173+
// 协程后，&file_ov 已经被系统 API给引用，防止野指针问题.
174+
back_readLength = get_overlapped_result(read_ov);
175+
// 现在，可以安全的执行 co_return -1 退出协程了.
176+
}
177+
178+
exit(1);
179+
}
180+
181+
if (read_file_pending)
182+
{
183+
back_readLength = get_overlapped_result(read_ov);
184+
}
185+
readLength = back_readLength;
186+
std::swap(pri_buf, back_buf);
187+
}
188+
//
189+
// All done. There is no need to call VirtualFree() to free CopyChunk
190+
// buffers here. The buffers will be freed when this process exits.
191+
//
192+
193+
CloseHandle(SourceFile);
194+
CloseHandle(DestFile);
195+
196+
//
197+
// We need another handle to the destination file that is
198+
// opened without FILE_FLAG_NO_BUFFERING. This allows us to set
199+
// the end-of-file marker to a position that is not sector-aligned.
200+
//
201+
HANDLE BufferedHandle = CreateFileA(destfilename.c_str(),
202+
GENERIC_WRITE,
203+
FILE_SHARE_READ | FILE_SHARE_WRITE,
204+
NULL,
205+
OPEN_EXISTING,
206+
0,
207+
NULL);
208+
if (BufferedHandle == INVALID_HANDLE_VALUE) {
209+
fprintf(stderr,
210+
"failed to open buffered handle to %s, error %d\n",
211+
destfilename.c_str(),
212+
GetLastError());
213+
exit(1);
214+
}
215+
216+
217+
//
218+
// Set the destination's file size to the size of the
219+
// source file, in case the size of the source file was
220+
// not a multiple of the page size.
221+
//
222+
Status = SetFilePointer(BufferedHandle,
223+
FileSize.LowPart,
224+
(PLONG)&FileSize.HighPart,
225+
FILE_BEGIN);
226+
if ((Status == INVALID_SET_FILE_POINTER) && (GetLastError() != NO_ERROR)) {
227+
fprintf(stderr, "final SetFilePointer failed, error %d\n", GetLastError());
228+
exit(1);
229+
}
230+
Success = SetEndOfFile(BufferedHandle);
231+
if (!Success) {
232+
fprintf(stderr, "SetEndOfFile failed, error %d\n", GetLastError());
233+
exit(1);
234+
}
235+
CloseHandle(BufferedHandle);
236+
237+
auto EndTime = std::chrono::steady_clock::now();
238+
239+
auto duration = EndTime - StartTime;
240+
auto dur_sec = std::chrono::duration_cast<std::chrono::milliseconds>(duration).count() / 1000.0f;
241+
242+
printf("%llu bytes copied in %.3f seconds\n", FileSize.QuadPart, dur_sec);
243+
printf("%.2f MB/sec\n", ((LONGLONG)FileSize.QuadPart/(1024.0*1024.0)) / dur_sec);
244+
245+
246+
create_detached_coroutine(std::bind(&copy_coroutine, IoPort, sourcefilename, destfilename));
247+
//exit(0);
248+
}
249+
250+
251+
int main(int argc, char* argv[])
252+
{
253+
254+
//
255+
// I/O completion port. All read and writes to the files complete
256+
// to this port.
257+
//
258+
HANDLE IoPort;
259+
260+
if (argc != 3) {
261+
fprintf(stderr, "Usage: %s SourceFile DestinationFile\n", argv[0]);
262+
exit(1);
263+
}
264+
265+
#ifdef _WIN32
266+
SYSTEM_INFO SystemInfo;
267+
GetSystemInfo(&SystemInfo);
268+
PageSize = SystemInfo.dwPageSize;
269+
#else
270+
PageSize = 4096;
271+
#endif
272+
273+
//
274+
//we are running on NT 3.51 or greater.
275+
//Create the I/O Completion Port
276+
//
277+
IoPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE,//file handle to associate with I/O completion port
278+
NULL, //optional handle to existing I/O completion port
279+
0, //completion key
280+
1); //# of threads allowed to execute concurrently
281+
282+
283+
if (IoPort == NULL) {
284+
fprintf(stderr, "failed to create ReadPort, error %d\n", GetLastError());
285+
exit(1);
286+
}
287+
288+
//
289+
// Do the copy
290+
//
291+
create_detached_coroutine(std::bind(&copy_coroutine, IoPort, std::string(argv[1]), std::string(argv[2])));
292+
293+
iocp::run_event_loop(IoPort);
294+
}

‎uasync/include/universal_fiber.hpp‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -184,7 +184,11 @@ struct FiberCallableArgument
184184
alignas(64) char callable_storage[sizeof(Callable)];
185185
};
186186

187-
inline constexpr std::size_t FiberContextSize = 65536;
187+
#ifndef COROTINE_STACK_SIZE
188+
#define COROTINE_STACK_SIZE 65536
189+
#endif
190+
191+
inline constexpr std::size_t FiberContextSize = COROTINE_STACK_SIZE;
188192

189193
template<typename Callable>
190194
struct FiberContext

0 commit comments

Comments
 (0)