Homework/03 async grpc

docs/01-basic/report.md

@@ -0,0 +1,39 @@
+# 01-basic 作业报告
+
+## Q1.1
+
+### Q1.1.1：哪条语句或哪几条语句将日志按逗号进行分割？代码中，我们是如何指定每一行的第几个字段代表何种意义的？
+
+(1)`LogFileParser.Parse(TextReader logFile)` 中，`using var csv = new CsvReader(logFile, config);` 创建 CsvHelper 的 CSV 读取器，`csv.GetRecords<LogRecord>()` 逐行读取并按 CSV 规则分列为 `LogRecord`。
+
+(2)`LogRecordMap` 的构造方法使用 `Map(m => m.LineNo).Index(0)`、`Map(m => m.Timestamp).Index(1)`、`Map(m => m.PodName).Index(2)`、`Map(m => m.Message).Index(3)`。因此第 0、1、2、3 列依次表示行号、时间戳、Pod 名和 JSON message。
+
+### Q1.1.2：在对日志中 JSON 格式的 message 字段进行读取时，我们是在哪个方法内用哪几条语句判断这一行日志的种类（Call / Request / Internal）的？
+
+`LineParser.ParseLine(LogRecord logRecord)` 中，先以 `JsonDocument.Parse(logRecord.Message)` 取得 JSON，再用 `root.TryGetProperty("event", out var eventElement)` 读取 `event`，最后用 `eventElement.GetString() switch` 分别选择 `CreateCall`、`CreateRequest`、`CreateInternal`。
+
+### Q1.1.3：在确定了日志种类后，我们是调用了哪个库方法对 JSON 进行解析的？进一步，我们的框架代码是如何防止日志中有字段缺失的？（例如所给的 Call 日志的 message 中缺失 request_id 字段）更进一步，日志中的 JSON 的键是 abc-def 命名法（称为烤串命名法），而我们的解析结果却是放在 AbcDef 命名法（称为大驼峰命名法）的属性里，我们的框架代码中是如何告诉 JSON 解析器完成这一命名法转换的？
+
+(1)`CreateCall`、`CreateRequest`、`CreateInternal` 分别调用 `JsonSerializer.Deserialize<CallMessage>()`、 `JsonSerializer.Deserialize<RequestMessage>()`、 `JsonSerializer.Deserialize<InternalMessage>()`。
+
+(2)`LineParser` 类末尾的三个 Message record 将必填属性标注为 `[property: JsonRequired]`；缺少相应属性时 `JsonSerializer.Deserialize` 会抛出 `JsonException`。
+
+(3)`LineParser` 类中的 `options` 字段设置了 `PropertyNamingPolicy = JsonNamingPolicy.KebabCaseLower`，因此 JSON 的 `request-id`、`status-code` 可映射至 C# 的 `RequestId`、`StatusCode`。
+
+## Q1.2：以一个 Call 事件的解析结果为例，当调用 KeyValueVisitor 的 Dump 方法后，都有哪些方法被调用？请补充完整如下的方法调用链.
+
+以 Call 类型日志为例，完整调用链为：
+1. `Dictionary<string, string> KeyValueVisitor.Dump(LogEntry entry)`  
+2. `TResult CallLogEntry.Accept<TResult>(ILogEntryVisitor<TResult> visitor)`  
+3. `Dictionary<string, string> KeyValueVisitor.Visit(CallLogEntry entry)`  
+
+第 1 步传入的是运行时类型为 `CallLogEntry` 的对象；第 2 步的
+`visitor.Visit(this)` 因而调用第 3 步的 Call 重载并返回字典。
+
+## Q1.3 b：如果使用了 AI，你给予 AI 的提示词是什么？你认为 AI 给出的解答、你完全凭借传统搜索引擎以及自己的能力能够写出的解答之间，AI 的解答比你好在哪？AI 又有哪些解答是存在问题的，或者至少是不如你自己的解答的？给出你的理由。
+
+(1)提示词：请根据本讲讲义内容以及作业题干，提取相关代码文件并将相关知识点标注在代码的注释中，并给出几个报告问题的讲解。（大意，经过多次迭代；代码暂未利用AI）
+
+(2)好处：能够帮助找到讲义中未讲解清楚或未涉及的知识点，理清回答思路，更清晰地掌握相应知识点。
+
+(3)问题：在问答轮数较少时容易偏离问题本身或采用不符合要求的方法。

docs/02-multithreading/report.md

@@ -0,0 +1,51 @@
+# 02-multithreading 作业报告
+
+## 已实现的功能
+
+我实现了 `WorkQueue<T>` 的多消费者安全队列；`LogFileAnalyzer` 的目录扫描、多线程并行解析、解析结果缓存和失败记录；以及 `LocalCli` 的目录切换、文件列表、指定/全部解析和结果查看功能。
+
+### 正常功能截图
+
+![完整功能截图1](./assets/normal-cli1.png)
+![完整功能截图2](./assets/normal-cli2.png)
+![完整功能截图3](./assets/normal-cli3.png)
+![完整功能截图4](./assets/normal-cli4.png)
+![完整功能截图5](./assets/normal-cli5.png)
+
+截图中演示了输入目录、显示日志文件、分析指定文件、分析全部文件、查看成功结果等所有常规功能。
+
+### 鲁棒性测试截图
+
+![鲁棒性测试截图1](./assets/invalid-input-cli1.png)
+![鲁棒性测试截图2](./assets/invalid-input-cli2.png)
+![鲁棒性测试截图3](./assets/invalid-input-cli3.png)
+![鲁棒性测试截图4](./assets/invalid-input-cli4.png)
+
+截图中演示了不存在或非法目录、非数字菜单输入、无效菜单选项、空文件名、当前目录不存在的文件名、负数并发度或非数字并发度等输入的报错处理。程序均给出提示并继续运行，没有崩溃。
+
+## Q2.1：我们把访问临界资源的程序片段称作临界区。在我们的多线程程序当中，临界资源即为不同线程的共享变量。请问：WorkQueue<T> 类中的共享变量有哪些？是通过什么保护其免于数据竞争（data race）呢？LogFileAnalyzer 类中的共享变量有哪些？是通过什么保护其免于数据竞争呢？如果条件变量的判断条件使用了 if 判断而非 while 判断，当出现了虚假唤醒现象时（在类 UNIX 系统中，由于 UNIX 信号等机制，即使没有人调用过 signal 或 broadcast，处于 wait 当中的条件变量也可能被唤醒），会出现什么后果？结合无限仓库容量的生产者消费者问题简单叙述一下。
+
+(1)`WorkQueue<T>` 的共享变量是 `_items`（内部 `Queue<T>`）和 `_isCompleted`（是否已结束加入）。
+(2)它们都在 `lock (_items)` 的临界区内读取或修改，因此同一时刻只有一个线程能访问这两项状态。消费者没有元素且尚未完成时使用 `Monitor.Wait(_items)` 暂时释放锁；生产者 `Enqueue` 后 `Monitor.Pulse(_items)` 唤醒一个等待者；`CompleteAdding` 修改完成标记后使用 `Monitor.PulseAll(_items)` 唤醒全部等待者。
+
+(3)`LogFileAnalyzer` 的共享变量包括 `_currentDirectory`、`_isAnalyzing`、`_logFiles` 和 `_analysisResults`。
+(4)它们由同一把 `_syncRoot` 锁保护。特别地，`AnalyzeFiles` 在锁内检查并设置 `_isAnalyzing`，防止两个调用同时开始分析；工作线程解析完文件后，也在 `lock (_syncRoot)` 中写入 `_analysisResults`。`finally` 中同样在锁内把 `_isAnalyzing` 复位为 `false`。
+
+(5)条件变量可能发生虚假唤醒，所以被唤醒并不保证“队列中已经有元素”。如果消费者只使用 `if (buffer == 0)` 调用 `Wait`，被虚假唤醒后会直接执行 `Dequeue`；在无限仓库容量的生产者消费者问题中，这会在仓库仍为空时尝试取出不存在的商品，导致异常或错误状态。使用 `while` 可以在每次唤醒后重新检查“队列是否为空、是否已经结束”这两个真实条件，因此正确处理虚假唤醒和多个消费者竞争。
+
+## Q2.2：在给出的代码框架 LogFileAnalyzer 中：哪一段代码扫描了给定的目录中的全部 .log 后缀的日志文件？假使给定的需求是不但要扫描给定目录中的日志文件，还要递归地获取给定的目录的全部子目录、子子目录……内的日志文件，应当如何做（简要回答即可）？
+
+(1)框架在 `ChangeDirectory` 中通过下列代码扫描当前目录的 `.log` 文件：
+
+```csharp
+Directory.EnumerateFiles(directoryPath, "*.log", SearchOption.TopDirectoryOnly)
+```
+
+它随后使用 `Select(Path.GetFileName)` 获取文件名、`OrderBy` 排序，并写入 `_logFiles` 与 `_analysisResults`。
+
+(2)若需要递归扫描当前目录及所有子目录，应将 `SearchOption.TopDirectoryOnly` 改为 `SearchOption.AllDirectories`。如果存在不同子目录下同名日志文件，还必须相应修改当前“以文件名为字典键”的设计，例如使用相对路径或完整路径作为键，以避免同名冲突。
+
+## Q2.3
+### Q2.3.b：如果使用了 AI，你给予 AI 的提示词是什么？你对 AI 的使用是询问 AI 一些接口的用法或是在某处的写法，还是让 AI 帮你写一部分作业代码，又或是让 AI 给你讲解代码框架？AI 的解答是否出现过错误（如果有，是哪些）？你认为本节的难度是偏低、适中，还是偏高？
+
+我使用了 AI。提示词大意为“请为我提供一版在原代码基础上通过注释标明用到的讲义中知识点和用途，并讲解几个todo部分代码的编写思路（如用到的类或语法等）”。我用它来查询接口、理解框架和理清思路，没有把未验证的回答直接当作结果。它曾出现的错误有破坏原框架等。本节难度为偏高。

docs/03-async-grpc/report.md

@@ -0,0 +1,33 @@
+# 03 async-gRPC 作业报告
+
+## T3.2 实现说明
+
+我实现了远程控制台客户端：切换日志目录、显示日志文件、按指定并行度分析指定文件或全部文件、流式查看分析结果。客户端全部使用 gRPC 的异步调用.
+
+## 完整功能截图
+
+![完整功能截图1](./assets/remote-cli1.png)
+![完整功能截图2](./assets/remote-cli2.png)
+![完整功能截图3](./assets/remote-cli3.png)
+![完整功能截图4](./assets/remote-cli4.png)
+![完整功能截图5](./assets/remote-cli5.png)
+![完整功能截图6](./assets/remote-cli6.png)
+
+## 鲁棒性测试截图
+
+我测试了菜单非法输入、目录错误后的恢复、负并行度/空文件名的本地校验、错误文件的服务端状态，以及“未选择目录直接分析”等异常处理。
+
+![鲁棒性截图1](./assets/remote-cli-robustness1.png)
+![鲁棒性截图2](./assets/remote-cli-robustness2.png)
+![鲁棒性截图3](./assets/remote-cli-robustness3.png)
+![鲁棒性截图4](./assets/remote-cli-robustness4.png)
+![鲁棒性截图5](./assets/remote-cli-robustness5.png)
+
+## Q3.1：你认为，你在开发网络应用程序，与你在以往开发非网络应用程序的区别在哪里？网络应用程序的开发存在哪些额外的难点？存在哪些额外的复杂之处？
+
+新增的难点主要在于：1.涉及不同端的连接和协同，程序结构更加复杂，构思和编写难度增加。2.需要处理的可能异常情形增加了连接失败等，鲁棒性维护难度增加。3.程序调试检查需要同时运行不同端，更不容易发现、定位和修正bug。
+
+## Q3.2
+### Q3.2.b：如果使用了 AI，你给予 AI 的提示词是什么？你对 AI 的使用是询问 AI 一些接口的用法、gRPC 的使用，或是在某处的写法，还是让 AI 帮你写一部分作业代码，又或是让 AI 给你讲解代码框架？AI 的解答是否出现过错误（如果有，是哪些）？你从 AI 那里是否得知了一些关于异步，或是 gRPC 等原本你不知道或是难以理解的知识？
+
+我使用了 AI。提示词大意为“请为我提供一版在原代码基础上通过注释标明用到的gRPC讲义中知识点和用途，并讲解几个todo部分代码的编写思路（如用到的类或语法等）的说明文档”。用途包括从复杂程序中寻找用到的接口、讲解用到的gRPC使用的知识点，以及提供代码编写思路和示例。AI的错误有类似功能代码写法不同导致混乱、弄混用到的相对地址和绝对地址等。得知的知识有领域模型与Proto模型的转换、gRPC各端安全与鲁棒性的维护方法等。

src/LocalCli/Program.cs

@@ -1,4 +1,5 @@
-using LogAnalyzer;
+using Google.Protobuf.WellKnownTypes;
+using LogAnalyzer;
 using LogParser.Visitors;
 
 namespace LocalCli
@@ -112,22 +113,72 @@ 6. Exit.
 
         private static void ShowLogFiles(LogFileAnalyzer analyzer)
         {
-            throw new NotImplementedException("T2.3");
+            foreach (var fileName in analyzer.GetLogFiles())
+            {
+                Console.WriteLine(fileName);
+            }
         }
 
         private static void AnalyzeFiles(LogFileAnalyzer analyzer)
         {
-            throw new NotImplementedException("T2.3");
+            Console.WriteLine("Please input file names separated by commas:");
+            var input = Console.ReadLine();
+            if (input is null)
+                return;
+            var fileNames = input.Split(',').Select(name => name.Trim());
+            Console.WriteLine("Please input degree of parallelism:");
+            try
+            {
+                analyzer.AnalyzeFiles(int.Parse(Console.ReadLine() ?? string.Empty), fileNames);
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine($"Analyze failed: {ex.Message}");
+            }
         }
 
         private static void AnalyzeAll(LogFileAnalyzer analyzer)
         {
-            throw new NotImplementedException("T2.3");
+            Console.WriteLine("Please input degree of parallelism:");
+            try
+            {
+                analyzer.AnalyzeAll(int.Parse(Console.ReadLine() ?? string.Empty));
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine($"Analyze failed:  {ex.Message}");
+            }
         }
 
         private static void GetAnalysisResult(LogFileAnalyzer analyzer)
         {
-            throw new NotImplementedException("T2.3");
+            Console.WriteLine("Please input file name:");
+            var fileName = Console.ReadLine();
+            if (fileName is null)
+                return;
+            if(!analyzer.TryGetAnalysisResult(fileName, out var result) || result is null)
+            {
+                Console.WriteLine("File does not exist.");
+                return;
+            }
+            switch (result.State)
+            {
+                case AnalysisState.NotAnalyzed:
+                    Console.WriteLine("This file has not been analyzed yet.");
+                    break;
+                case AnalysisState.Succeeded:
+                    var visitor = new KeyValueVisitor();
+                    foreach (var entry in result.Entries)
+                    {
+                        var fields = visitor.Dump(entry);
+                        Console.WriteLine(string.Join(", ",fields.Select(
+                            field => $"{field.Key}={field.Value}")));
+                    }
+                    break;
+                case AnalysisState.Failed:
+                    Console.WriteLine($"Analysis failed: {result.ErrorMessage}");
+                    break;
+            }
         }
     }
 }

src/LogAnalyzer/LogFileAnalyzer.cs

@@ -137,11 +137,7 @@ public void AnalyzeFiles(int degreeOfParallelism, IEnumerable<string> fileNames)
                     }
                 }
                 fileList = fileNameList.Select(fileName => _logFiles[fileName]).ToList();
-
-                /*
-                 * Set _isAnalyzing
-                 */
-                // TODO: T2.2
+                _isAnalyzing = true;
             }
 
             try
@@ -150,11 +146,10 @@ public void AnalyzeFiles(int degreeOfParallelism, IEnumerable<string> fileNames)
             }
             finally
             {
-                /*
-                 * Unset _isAnalyzing
-                 * Remember to lock _syncRoot to prevent data race
-                 */
-                // TODO: T2.2
+                lock (_syncRoot)
+                {
+                    _isAnalyzing = false;
+                }
             }
         }
 
@@ -165,11 +160,14 @@ private void RunWorkers(int degreeOfParallelism, IReadOnlyList<FileInfo> fileLis
             {
                 foreach (var file in fileList)
                 {
-                    /*
-                     * Filter unparsed files.
-                     * If there is an unknown file, throw System.InvalidOperationException.
-                     */
-                    throw new NotImplementedException("TODO: T2.2");
+                    if (!_analysisResults.ContainsKey(file.Name))
+                    {
+                        throw new InvalidOperationException($"Unkown log file: {file.Name}");
+                    }
+                    if (_analysisResults[file.Name].State == AnalysisState.NotAnalyzed)
+                    {
+                        logFilesToParse.Add(file);
+                    }
                 }
             }
 
@@ -180,27 +178,29 @@ private void RunWorkers(int degreeOfParallelism, IReadOnlyList<FileInfo> fileLis
 
             var queue = new WorkQueue<FileInfo>();
 
-            /*
-             * Enqueue log files
-             */
-            // TODO: T2.2
-
+            foreach (var file in logFilesToParse)
+            {
+                queue.Enqueue(file);
+            }
+            queue.CompleteAdding();
+
             degreeOfParallelism = Math.Max(Math.Min(degreeOfParallelism, logFilesToParse.Count), 1);
             var workers = new Thread[degreeOfParallelism];
             for (int i = 0; i < degreeOfParallelism; i++)
             {
                 int workerId = i;
                 string threadName = $"log-analyzer-worker-{workerId}";
-                /*
-                 * Create and start threads to run `WorkerMain`
-                 */
-                // TODO: T2.2
+                workers[i] = new Thread(() => WorkerMain(workerId, queue))
+                {
+                    Name = threadName
+                };
+                workers[i].Start();
             }
 
-            /*
-             * Wait for (join) all threads to end
-             */
-            // TODO: T2.2
+            foreach (var worker in workers)
+            {
+                worker.Join();
+            }
         }
 
         private void WorkerMain(int workerId, WorkQueue<FileInfo> queue)
@@ -212,20 +212,31 @@ private void WorkerMain(int workerId, WorkQueue<FileInfo> queue)
                 AnalysisResult result;
                 try
                 {
-                    // Parse file
-                    throw new NotImplementedException("TODO: T2.2");
+                    using var reader = new StreamReader(file.FullName);
+                    var entries = parser.Parse(reader).ToList();
+                    result = new AnalysisResult(
+                        FileName: file.Name,
+                        FullName: file.FullName,
+                        State: AnalysisState.Succeeded,
+                        Entries: entries,
+                        ErrorMessage: null,
+                        WorkerId: workerId);
                 }
                 catch (Exception ex)
                 {
-                    // Save exception message to result
-                    throw new NotImplementedException("TODO: T2.2");
+                    result = new AnalysisResult(
+                        FileName: file.Name,
+                        FullName: file.FullName,
+                        State: AnalysisState.Failed,
+                        Entries: Array.Empty<LogEntry>(),
+                        ErrorMessage: ex.Message,
+                        WorkerId: workerId);
                 }
 
-                /*
-                 * Save parse result.
-                 * [!Important] Remember to lock _syncRoot to prevent data race.
-                 */
-                throw new NotImplementedException("TODO: T2.2");
+                lock (_syncRoot)
+                {
+                    _analysisResults[file.Name] = result;
+                }
             }
         }
     }