From 096465179265cffed65e2fab0e2359e878fab454 Mon Sep 17 00:00:00 2001 From: XieHan Date: Mon, 1 Aug 2022 22:37:48 +0800 Subject: [PATCH 01/11] Add WFRepeaterTask --- src/factory/WFTask.h | 65 ++++++++++++++++++++++++++++++++++- src/factory/WFTaskFactory.h | 9 +++++ src/factory/WFTaskFactory.inl | 4 +-- 3 files changed, 75 insertions(+), 3 deletions(-) diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index 6df92a74..4003cbf2 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -45,7 +45,7 @@ enum WFT_STATE_SSL_ERROR = 65, WFT_STATE_DNS_ERROR = 66, /* for client task only */ WFT_STATE_TASK_ERROR = 67, - WFT_STATE_ABORTED = CS_STATE_STOPPED /* main process terminated */ + WFT_STATE_ABORTED = CS_STATE_STOPPED }; template @@ -721,6 +721,69 @@ protected: virtual ~WFGoTask() { } }; +class WFRepeaterTask : public WFGenericTask +{ +public: + void set_create(std::function create) + { + this->create = std::move(create); + } + + void set_callback(std::function callback) + { + this->callback = std::move(callback); + } + +protected: + virtual void dispatch() + { + SubTask *task = this->create(this); + + if (task) + { + series_of(this)->push_front(this); + series_of(this)->push_front(task); + } + else + { + this->create = nullptr; + this->state = WFT_STATE_SUCCESS; + } + + this->subtask_done(); + } + + virtual SubTask *done() + { + SeriesWork *series = series_of(this); + + if (!this->create) + { + if (this->callback) + this->callback(this); + + delete this; + } + + return series->pop(); + } + +protected: + std::function create; + std::function callback; + +public: + WFRepeaterTask(std::function&& create, + std::function&& cb) : + create(std::move(create)), + callback(std::move(cb)) + { + } + +protected: + virtual ~WFRepeaterTask() { } +}; + class WFModuleTask : public ParallelTask, protected SeriesWork { public: diff --git a/src/factory/WFTaskFactory.h b/src/factory/WFTaskFactory.h index 9111a39f..42bbcc80 100644 --- a/src/factory/WFTaskFactory.h +++ b/src/factory/WFTaskFactory.h @@ -101,6 +101,9 @@ using WFEmptyTask = WFGenericTask; using WFDynamicTask = WFGenericTask; using dynamic_create_t = std::function; +using repeated_create_t = std::function; +using repeater_callback_t = std::function; + using module_callback_t = std::function; class WFTaskFactory @@ -314,6 +317,12 @@ public: static WFDynamicTask *create_dynamic_task(dynamic_create_t create); + static WFRepeaterTask *create_repeater_task(repeated_create_t create, + repeater_callback_t callback) + { + return new WFRepeaterTask(std::move(create), std::move(callback)); + } + public: static WFModuleTask *create_module_task(SubTask *first, module_callback_t callback) diff --git a/src/factory/WFTaskFactory.inl b/src/factory/WFTaskFactory.inl index f9b2d5bb..8a9b0227 100644 --- a/src/factory/WFTaskFactory.inl +++ b/src/factory/WFTaskFactory.inl @@ -145,8 +145,8 @@ protected: std::function create; public: - __WFDynamicTask(std::function&& func) : - create(std::move(func)) + __WFDynamicTask(std::function&& create) : + create(std::move(create)) { } }; From eb78a8bdf2c73b2c767d34002b5df940a43db5fc Mon Sep 17 00:00:00 2001 From: XieHan Date: Mon, 1 Aug 2022 22:48:24 +0800 Subject: [PATCH 02/11] update WFRepeaterTask --- src/factory/WFTask.h | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index 4003cbf2..a9e958c2 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -745,10 +745,7 @@ protected: series_of(this)->push_front(task); } else - { - this->create = nullptr; this->state = WFT_STATE_SUCCESS; - } this->subtask_done(); } @@ -757,7 +754,7 @@ protected: { SeriesWork *series = series_of(this); - if (!this->create) + if (this->state != WFT_STATE_UNDEFINED) { if (this->callback) this->callback(this); From 2b7f7aeb6b9456c12f72040aaab20df0c2392167 Mon Sep 17 00:00:00 2001 From: XieHan Date: Tue, 2 Aug 2022 02:29:05 +0800 Subject: [PATCH 03/11] Update tutorial-09-http_file_server with WFRepeaterTask --- tutorial/tutorial-09-http_file_server.cc | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/tutorial/tutorial-09-http_file_server.cc b/tutorial/tutorial-09-http_file_server.cc index 2babda0d..67dd96fd 100644 --- a/tutorial/tutorial-09-http_file_server.cc +++ b/tutorial/tutorial-09-http_file_server.cc @@ -132,9 +132,7 @@ int main(int argc, char *argv[]) } /* Test the server. */ - while (1) - { - WFFacilities::WaitGroup wg(1); + auto&& create = [&scheme, port](WFRepeaterTask *)->SubTask *{ char buf[1024]; *buf = '\0'; printf("Input file name: (Ctrl-D to exit): "); @@ -142,12 +140,12 @@ int main(int argc, char *argv[]) if (*buf == '\0') { printf("\n"); - break; + return nullptr; } std::string url = scheme + "127.0.0.1:" + std::to_string(port) + "/" + buf; WFHttpTask *task = WFTaskFactory::create_http_task(url, 0, 0, - [&wg](WFHttpTask *task) { + [](WFHttpTask *task) { auto *resp = task->get_resp(); if (strcmp(resp->get_status_code(), "200") == 0) { @@ -159,13 +157,19 @@ int main(int argc, char *argv[]) { printf("%s %s\n", resp->get_status_code(), resp->get_reason_phrase()); } - - wg.done(); }); - task->start(); - wg.wait(); - } + return task; + }; + + WFFacilities::WaitGroup wg(1); + WFRepeaterTask *repeater; + repeater = WFTaskFactory::create_repeater_task(create, [&wg](WFRepeaterTask *) { + wg.done(); + }); + + repeater->start(); + wg.wait(); server.stop(); return 0; From 9706ef25f451d90942a8e33c1ba5bdece7e6abeb Mon Sep 17 00:00:00 2001 From: XieHan Date: Tue, 2 Aug 2022 02:54:31 +0800 Subject: [PATCH 04/11] Update tutorial-10's client with WFRepeaterTask --- .../client.cc | 70 +++++++++---------- 1 file changed, 34 insertions(+), 36 deletions(-) diff --git a/tutorial/tutorial-10-user_defined_protocol/client.cc b/tutorial/tutorial-10-user_defined_protocol/client.cc index eb15762a..943f3f42 100644 --- a/tutorial/tutorial-10-user_defined_protocol/client.cc +++ b/tutorial/tutorial-10-user_defined_protocol/client.cc @@ -59,51 +59,49 @@ int main(int argc, char *argv[]) host = argv[1]; port = atoi(argv[2]); - std::function callback = - [&host, port, &callback](WFTutorialTask *task) { - int state = task->get_state(); - int error = task->get_error(); - TutorialResponse *resp = task->get_resp(); + + auto&& create = [host, port](WFRepeaterTask *)->SubTask *{ char buf[1024]; - void *body; - size_t body_size; - - if (state != WFT_STATE_SUCCESS) - { - if (state == WFT_STATE_SYS_ERROR) - fprintf(stderr, "SYS error: %s\n", strerror(error)); - else if (state == WFT_STATE_DNS_ERROR) - fprintf(stderr, "DNS error: %s\n", gai_strerror(error)); - else - fprintf(stderr, "other error.\n"); - return; - } - - resp->get_message_body_nocopy(&body, &body_size); - if (body_size != 0) - printf("Server Response: %.*s\n", (int)body_size, (char *)body); - printf("Input next request string (Ctrl-D to exit): "); *buf = '\0'; scanf("%1023s", buf); - body_size = strlen(buf); - if (body_size > 0) + size_t body_size = strlen(buf); + if (body_size == 0) { - WFTutorialTask *next; - next = MyFactory::create_tutorial_task(host, port, 0, callback); - next->get_req()->set_message_body(buf, body_size); - next->get_resp()->set_size_limit(4 * 1024); - **task << next; /* equal to: series_of(task)->push_back(next) */ - } - else printf("\n"); + return nullptr; + } + + WFTutorialTask *task = MyFactory::create_tutorial_task(host, port, 0, + [](WFTutorialTask *task) { + int state = task->get_state(); + int error = task->get_error(); + TutorialResponse *resp = task->get_resp(); + void *body; + size_t body_size; + + if (state == WFT_STATE_SUCCESS) + { + resp->get_message_body_nocopy(&body, &body_size); + printf("Server Response: %.*s\n", (int)body_size, (char *)body); + } + else + { + const char *str = WFGlobal::get_error_string(state, error); + fprintf(stderr, "Error: %s\n", str); + } + }); + + task->get_req()->set_message_body(buf, body_size); + task->get_resp()->set_size_limit(4 * 1024); + return task; }; - /* First request is emtpy. We will ignore the server response. */ WFFacilities::WaitGroup wait_group(1); - WFTutorialTask *task = MyFactory::create_tutorial_task(host, port, 0, callback); - task->get_resp()->set_size_limit(4 * 1024); - Workflow::start_series_work(task, [&wait_group](const SeriesWork *) { + + WFRepeaterTask *repeater; + repeater = WFTaskFactory::create_repeater_task(std::move(create), nullptr); + Workflow::start_series_work(repeater, [&wait_group](const SeriesWork *) { wait_group.done(); }); From 4fa4186d46f4032196abfb00be8a441501ad0383 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Tue, 2 Aug 2022 19:35:40 +0800 Subject: [PATCH 05/11] Update tutorial-10-user_defined_protocol.md --- docs/en/tutorial-10-user_defined_protocol.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/docs/en/tutorial-10-user_defined_protocol.md b/docs/en/tutorial-10-user_defined_protocol.md index d97e1e83..112daca1 100644 --- a/docs/en/tutorial-10-user_defined_protocol.md +++ b/docs/en/tutorial-10-user_defined_protocol.md @@ -202,8 +202,7 @@ Please see [server.cc](/tutorial/tutorial-10-user_defined_protocol/server.cc) fo # client -The logic of the client is to receive the user input from standard IO, construct a request, send it to the server and get the results. -For simplicity, the process of reading standard input is completed in the callback, so we will send an empty request first. Also, for the sake of security, we limit the packet size of the server reply to 4KB. +The logic of the client is to receive the user input from standard IO, construct a request, send it to the server and get the results. Here we use WFRepeaterTask to implement the repeating process. For the sake of security, we limit the packet size of the server reply to 4KB. The only thing that a client needs to know is how to generate a client task on a user-defined protocol. There are three interface options in [WFTaskFactory.h](/src/factory/WFTaskFactory.h): ~~~cpp From bae79a1257ad89309ccba41c5bf6956450bdb17a Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Tue, 2 Aug 2022 19:39:20 +0800 Subject: [PATCH 06/11] Update tutorial-10-user_defined_protocol.md --- docs/tutorial-10-user_defined_protocol.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/tutorial-10-user_defined_protocol.md b/docs/tutorial-10-user_defined_protocol.md index 94563b4c..306b9a26 100644 --- a/docs/tutorial-10-user_defined_protocol.md +++ b/docs/tutorial-10-user_defined_protocol.md @@ -191,8 +191,8 @@ server与普通的http server没有什么区别。我们优先IPv6启动，这 # client端 -client端的逻辑是从标准IO接收用户输入，构造出请求发往server并得到结果。 -为了简单，读取标准输入的过程都在callback里完成，因此我们会先发出一条空请求。同样为了安全我们限制server回复包不超4KB。 +client端的逻辑是从标准IO接收用户输入，构造出请求发往server并得到结果。这里我们使用了WFRepeaterTask来实现这个重复过程，直到用户的输入为空。 +此外，为了安全我们限制server回复包不超4KB。 client端唯一需要了解的就是怎么产生一个自定义协议的client任务，在[WFTaskFactory.h](../src/factory/WFTaskFactory.h)有三个接口可以选择： ~~~cpp template From 034fc490656aceda6f76b9f93cdf17f068d49259 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Tue, 2 Aug 2022 19:41:12 +0800 Subject: [PATCH 07/11] Update tutorial-10-user_defined_protocol.md --- docs/en/tutorial-10-user_defined_protocol.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/en/tutorial-10-user_defined_protocol.md b/docs/en/tutorial-10-user_defined_protocol.md index 112daca1..fc44ccf8 100644 --- a/docs/en/tutorial-10-user_defined_protocol.md +++ b/docs/en/tutorial-10-user_defined_protocol.md @@ -202,7 +202,7 @@ Please see [server.cc](/tutorial/tutorial-10-user_defined_protocol/server.cc) fo # client -The logic of the client is to receive the user input from standard IO, construct a request, send it to the server and get the results. Here we use WFRepeaterTask to implement the repeating process. For the sake of security, we limit the packet size of the server reply to 4KB. +The logic of the client is to receive the user input from standard IO, construct a request, send it to the server and get the results. Here we use WFRepeaterTask to implement the repeating process, terminates if the user's input is empty. For the sake of security, we limit the packet size of the server reply to 4KB. The only thing that a client needs to know is how to generate a client task on a user-defined protocol. There are three interface options in [WFTaskFactory.h](/src/factory/WFTaskFactory.h): ~~~cpp From 120507fd5c04c108c80e27233e964e4f32ccede1 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Tue, 2 Aug 2022 20:20:02 +0800 Subject: [PATCH 08/11] Update tutorial-09-http_file_server.md --- docs/tutorial-09-http_file_server.md | 48 ++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/docs/tutorial-09-http_file_server.md b/docs/tutorial-09-http_file_server.md index eb71b47c..02fe891e 100644 --- a/docs/tutorial-09-http_file_server.md +++ b/docs/tutorial-09-http_file_server.md @@ -7,6 +7,7 @@ http_file_server是一个web服务器，用户指定启动端口，根路径（默认为程序当路程），就可以启动一个web server。 用户还可以指定一个PEM格式的certificate file和key file，启动一个https web server。 +程序在启动server之后，可以从命令行接受用户输入，并通过127.0.0.1地址来访问这个server。 程序主要展示了磁盘IO任务的用法。在Linux系统下，我们利用了Linux底层的aio接口，文件读取完全异步。 # 启动server @@ -141,6 +142,53 @@ buf域的内存我们是自己管理的，可以通过append_output_body_nocopy( 在回复完成后，我们会free()这块内存，这个语句在process里： server_task->set_callback([](WFHttpTask *t){ free(t->user_data); }); +# 命令行交互 + +启动server后，用户可以在控制台输入文件名来访问server。当输入文件名为空（Ctrl-D），关闭server并结束程序。 +这里，我们使用了WFRepeaterTask来实现这个循环接受输入的过程。WFRepeaterTask是一种循环任务，产生的接口如下： +~~~cpp +using repeated_create_t = std::function; +using repeater_callback_t = std::function; + +class WFTaskFactory +{ + WFRpeaterTask *create_repeater_task(repeated_create_t create, repeater_callback_t callback); +}; +~~~ +通过create函数，可以创建一个repeater任务。repeater内部会反复调用create，产生一个任务并运行，直到create返回空指针。 +在我们的这个示例里，create函数内部调用scanf。当用户输入为空时，create返回NULL，整个循环过程结束。 +当用户输入不为空（文件名），产生一个访问127.0.0.1地址的http任务来访问我们开启的server。 +~~~cpp +{ + auto&& create = [&scheme, port](WFRepeaterTask *)->SubTask *{ + ... + scanf("%1023s", buf); + if (*buf == '\0') + return NULL; + + std::string url = scheme + "127.0.0.1:" + std::to_string(port) + "/" + buf; + WFHttpTask *task = WFTaskFactory::create_http_task(url, 0, 0, + [](WFHttpTask *task) { + ... + }); + + return task; + }; + + WFFacilities::WaitGroup wg(1); + WFRepeaterTask *repeater; + repeater = WFTaskFactory::create_repeater_task(create, [&wg](WFRepeaterTask *) { + wg.done(); + }); + + repeater->start(); + wg.wait(); + + server.stop(); +} +~~~ +最后，当create返回NULL，repeater被callback。我们关闭server并结束程序。 + # 关于文件异步IO的实现 Linux操作系统支持一套效率很高，CPU占用非常少的异步IO系统调用。在Linux系统下使用我们的框架将默认使用这套接口。 From 96d7a172f50fbd51da9644ae2ddd9ce43b0a7666 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Tue, 2 Aug 2022 20:20:40 +0800 Subject: [PATCH 09/11] Update tutorial-09-http_file_server.cc --- tutorial/tutorial-09-http_file_server.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tutorial/tutorial-09-http_file_server.cc b/tutorial/tutorial-09-http_file_server.cc index 67dd96fd..210804a2 100644 --- a/tutorial/tutorial-09-http_file_server.cc +++ b/tutorial/tutorial-09-http_file_server.cc @@ -140,7 +140,7 @@ int main(int argc, char *argv[]) if (*buf == '\0') { printf("\n"); - return nullptr; + return NULL; } std::string url = scheme + "127.0.0.1:" + std::to_string(port) + "/" + buf; From 2ee5e85af0a761a300e559a549715b1f9879b2b9 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Tue, 2 Aug 2022 20:21:03 +0800 Subject: [PATCH 10/11] Update client.cc --- tutorial/tutorial-10-user_defined_protocol/client.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tutorial/tutorial-10-user_defined_protocol/client.cc b/tutorial/tutorial-10-user_defined_protocol/client.cc index 943f3f42..46724637 100644 --- a/tutorial/tutorial-10-user_defined_protocol/client.cc +++ b/tutorial/tutorial-10-user_defined_protocol/client.cc @@ -69,7 +69,7 @@ int main(int argc, char *argv[]) if (body_size == 0) { printf("\n"); - return nullptr; + return NULL; } WFTutorialTask *task = MyFactory::create_tutorial_task(host, port, 0, From ba7e9e627eaf0c405114fa2576cfafb4add616a6 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Tue, 2 Aug 2022 20:38:57 +0800 Subject: [PATCH 11/11] Update tutorial-09-http_file_server.md --- docs/en/tutorial-09-http_file_server.md | 46 ++++++++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/docs/en/tutorial-09-http_file_server.md b/docs/en/tutorial-09-http_file_server.md index 456e3216..e329f886 100644 --- a/docs/en/tutorial-09-http_file_server.md +++ b/docs/en/tutorial-09-http_file_server.md @@ -7,7 +7,7 @@ # About http\_file\_server http\_file\_server is a web server. You can start a web server after specifying the startup port and the root path (the default setting is the current path). -You can also specify a certificate file and a key file in PEM format to start an HTTPS web server. +You can also specify a certificate file and a key file in PEM format to start an HTTPS web server. User may access the server through command line, the request will be sent to IP address 127.0.0.1. The program mainly demonstrates how to use disk IO tasks. In the Linux system, we use the aio interface in the kernel of Linux, and the file reading is completely asynchronous. # Starting a server @@ -149,6 +149,50 @@ The memory of the buf domain is managed by ourselves. You can use **append\_out After the reply is completed, we will **free()** this block of memory with this line in the process: server\_task->set\_callback(\[](WFHttpTask \*t){ free(t->user\_data); }); +# Interact with the server through command line + +After the server is started, users may access it through command line. Simply input the file name that you want to get, or input Ctrl-D to end the program. The repeating process is implemnted by using WFRepeaterTask, which can be created by this factory function: +~~~cpp +using repeated_create_t = std::function; +using repeater_callback_t = std::function; + +class WFTaskFactory +{ + WFRpeaterTask *create_repeater_task(repeated_create_t create, repeater_callback_t callback); +}; +~~~ +As above, a repeater task is created with a task creator function. The repeater calls the task creator repeatedly and run the task until the creator return a NULL pointer. When the using's input is not empty, our creator will create an HTTP task on IP 127.0.0.1 to access the server. +~~~cpp +{ + auto&& create = [&scheme, port](WFRepeaterTask *)->SubTask *{ + ... + scanf("%1023s", buf); + if (*buf == '\0') + return NULL; + + std::string url = scheme + "127.0.0.1:" + std::to_string(port) + "/" + buf; + WFHttpTask *task = WFTaskFactory::create_http_task(url, 0, 0, + [](WFHttpTask *task) { + ... + }); + + return task; + }; + + WFFacilities::WaitGroup wg(1); + WFRepeaterTask *repeater; + repeater = WFTaskFactory::create_repeater_task(create, [&wg](WFRepeaterTask *) { + wg.done(); + }); + + repeater->start(); + wg.wait(); + + server.stop(); +} +~~~ +Finally, when the creator returned NULL, the repeater's callback is called and the program will be ended. + # About the implementation of the file IO Linux operating system supports a set of asynchronous IO system calls with high efficiency and very little CPU occupation. If you use our framework in a Linux system, this set of interfaces are used by default.