Homework/02 multithreading

docs/01-basic/report.md

@@ -0,0 +1,73 @@
+### (Q1.1)
+
+在给出的代码框架 `Parser` 中：
+
++ 哪条语句或哪几条语句将日志按逗号进行分割？代码中，我们是如何指定每一行的第几个字段代表何种意义的？
+```
+var config = new CsvConfiguration(CultureInfo.InvariantCulture)
+{
+    HasHeaderRecord = false
+};
+using var csv = new CsvReader(logFile, config);
+csv.Context.RegisterClassMap<LogRecordMap>();
+
+foreach (var logRecord in csv.GetRecords<LogRecord>())
+{
+    yield return LineParser.ParseLine(logRecord);
+}
+```
+```
+internal class LogRecordMap : ClassMap<LogRecord>
+{
+    public 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);
+    }
+}
+```
+
++ 在对日志中 JSON 格式的 `message` 字段进行读取时，我们是在哪个方法内用哪几条语句判断这一行日志的种类（Call / Request / Internal）的？
+```
+eventElement.GetString() switch
+{
+    "call" => LineParser.CreateCall(logRecord),
+    "request" => LineParser.CreateRequest(logRecord),
+    "internal" => LineParser.CreateInternal(logRecord),
+    _ => throw new FormatException($"Unknown event type: {eventElement.GetString()} in log message: {logRecord.Message}")
+};
+```
++ 在确定了日志种类后，我们是调用了哪个库方法对 JSON 进行解析的？
+
+`JsonSerializer.Deserialize`
+
++ 进一步，我们的框架代码是如何防止日志中有字段缺失的？（例如所给的 Call 日志的 `message` 中缺失 `request_id` 字段）
+
+`throw new FormatException(...)`
+
++ 更进一步，日志中的 JSON 的键是 `abc-def` 命名法（称为烤串命名法），而我们的解析结果却是放在 `AbcDef` 命名法（称为大驼峰命名法）的属性里，我们的框架代码中是如何告诉 JSON 解析器完成这一命名法转换的？
+
+```
+private static JsonSerializerOptions options = new JsonSerializerOptions
+{
+    PropertyNamingPolicy = JsonNamingPolicy.KebabCaseLower,
+};
+
+JsonSerializer.Deserialize<T>(..., options);
+```
+
+### (Q1.2)
+
+以一个 Call 事件的解析结果为例，当调用 `KeyValueVisitor` 的 `Dump` 方法后，都有哪些方法被调用？请补充完整如下的方法调用链（.NET 内置库无需写出）：
+
++ `Dictionary<string, string> KeyValueVisitor.Dump(LogEntry entry)`
++ `TResult Accept<TResult>(ILogEntryVisitor<TResult> visitor)`
++ `Dictionary<string, string> Visit(CallLogEntry entry)`
+
+### (Q1.3)
+
+未使用AI。从开始到通过全部测试花费1.5小时。本次作业相比程序设计作业代码量较少，具体逻辑编写也较为简单，但是需要理解已有代码架构具有一定难度。目前作答并不完美，有以下两个问题：
+1. 对于 internal message 解析的操作并未考虑可能的不存在 ':' 时的情况，此时代码将直接抛出异常
+2. 代码commit信息和此前风格未保持一致，未使用git emoji

docs/02-multithreading/report.md

@@ -0,0 +1,50 @@
+功能截图：
+![functions.png](./functions.png)
+
+鲁棒性测试：
+![robust.png](./robust.png)
+
+### (Q2.1)
+
+本问题考察关于临界区的理解。
+
+我们把访问临界资源的程序片段称作临界区。在我们的多线程程序当中，临界资源即为不同线程的共享变量。请问：
+
++ `WorkQueue<T>` 类中的共享变量有哪些？是通过什么保护其免于数据竞争（data race）呢？
+1. _items和_isCompleted
+2. 通过lock(_items)
++ `LogFileAnalyzer` 类中的共享变量有哪些？是通过什么保护其免于数据竞争呢？
+1. _currentDirectory, _isAnalyzing, _logFiles, _analysisResults
+2. lock(_syncRoot)
++ 如果条件变量的判断条件使用了 `if` 判断而非 `while` 判断，当出现了虚假唤醒现象时（在类 UNIX 系统中，由于 UNIX 信号等机制，即使没有人调用过 `signal` 或 `broadcast`，处于 `wait` 当中的条件变量也可能被唤醒），会出现什么后果？结合无限仓库容量的生产者消费者问题简单叙述一下。
+
+此时被唤醒的线程会误以为等待条件已经达成继续执行下面的代码。对于无限仓库容量的生产者消费者模型来说，消费者线程的等待条件通常是队列为空。当被虚假唤醒，消费者线程会尝试从队列中取出数据，但是实际上队列还是空的，所以可能直接抛出异常或者取出无效数据。
+
+### (Q2.2)
+
+在给出的代码框架 `LogFileAnalyzer` 中：
+
++ 那一段代码扫描了给定的目录中的全部 `.log` 后缀的日志文件？假使给定的需求是不但要扫描给定目录中的日志文件，还要递归地获取给定的目录的全部子目录、子子目录……内的日志文件，应当如何做（简要回答即可）？
+
+1. 
+```
+var logFiles = Directory.EnumerateFiles(directoryPath, "*.log", SearchOption.TopDirectoryOnly)
+```
+
+2. 
+```
+var logFiles = Directory.EnumerateFiles(directoryPath, "*.log", SearchOption.AllDirectories)
+```
+### (Q2.3)
+
+本次作业中，你是否使用了 AI？根据你的使用情况，在以下 (Q2.3.a) (Q2.3.b) 两个问题中选择一题作答：
+
+#### (Q2.3.a)
+
+如果没有使用 AI，你花了大约多长时间通过全部测试？你认为本次作业相比于你曾经上过的程序设计课程的作业难度如何？你是否借助了传统搜索引擎来完成本节？你认为本节的难度是偏低、适中，还是偏高？
+
+1. 未使用AI。
+2. 总共花费2.5小时完成。
+3. 相比程序设计课作业难度更高，主要在线程安全相关考虑上。
+4. 使用了搜索引擎，主要搜索一些内置API的使用方法。
+5. 难度适中。

src/LocalCli/Program.cs

@@ -112,22 +112,92 @@
 
         private static void ShowLogFiles(LogFileAnalyzer analyzer)
         {
-            throw new NotImplementedException("T2.3");
+            foreach (var file in analyzer.GetLogFiles())
+            {
+                Console.WriteLine(file);
+            }
         }
 
         private static void AnalyzeFiles(LogFileAnalyzer analyzer)
         {
-            throw new NotImplementedException("T2.3");
+            Console.WriteLine("Degree of parrallelism:");
+            var parrallelism = Console.ReadLine();
+            if (parrallelism == null)
+            {
+                return;
+            }
+            Console.WriteLine("Filenames, split with ',':");
+            var filesStr = Console.ReadLine();
+            if (filesStr == null)
+            {
+                return;
+            }
+            try
+            {
+                analyzer.AnalyzeFiles(int.Parse(parrallelism), filesStr.Split(',').Select(x => x.Trim()));
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine(ex.ToString());
+            }
         }
 
         private static void AnalyzeAll(LogFileAnalyzer analyzer)
         {
-            throw new NotImplementedException("T2.3");
+            Console.WriteLine("Degree of parrallelism:");
+            var input = Console.ReadLine();
+            if (input == null)
+            {
+                return;
+            }
+            int degreeOfParallelism;
+            try
+            {
+                degreeOfParallelism = int.Parse(input);
+                analyzer.AnalyzeAll(degreeOfParallelism);
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine(ex.ToString());
+            }
         }
 
         private static void GetAnalysisResult(LogFileAnalyzer analyzer)
         {
-            throw new NotImplementedException("T2.3");
+            Console.WriteLine("Filename:");
+            var filename = Console.ReadLine().Trim();
+            if (filename == null)
+            {
+                return;
+            }
+            if (analyzer.TryGetAnalysisResult(filename, out AnalysisResult? result))
+            {
+                if (result.State == AnalysisState.Succeeded)
+                {
+                    var visitor = new KeyValueVisitor();
+                    foreach (var entry in result.Entries)
+                    {
+                        var dumpedInfo = visitor.Dump(entry);
+                        foreach (var item in dumpedInfo)
+                        {
+                            Console.Write($"[{item.Key}] {item.Value} ");
+                        }
+                        Console.WriteLine();
+                    }
+                }
+                else if (result.State == AnalysisState.Failed)
+                {
+                    Console.WriteLine("Analysis failed");
+                }
+                else if (result.State == AnalysisState.NotAnalyzed)
+                {
+                    Console.WriteLine("Not analyzed yet");
+                }
+            }
+            else
+            {
+                Console.WriteLine($"No parse result for file {filename}.");
+            }
         }
     }
 }

src/LogAnalyzer/LogFileAnalyzer.cs

@@ -1,7 +1,5 @@
 using LogParser.Models;
 using LogParser.Parser;
-using System.Diagnostics.CodeAnalysis;
-using System.Security.Cryptography.X509Certificates;
 
 namespace LogAnalyzer
 {
@@ -138,10 +136,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 +145,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 +159,18 @@ 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");
+                    try
+                    {
+                        if (_analysisResults[file.Name].State != AnalysisState.NotAnalyzed)
+                        {
+                            continue;
+                        }
+                    }
+                    catch (KeyNotFoundException)
+                    {
+                        throw new InvalidOperationException($"Unknown file name {file.Name}");
+                    }
+                    logFilesToParse.Add(file);
                 }
             }
 
@@ -180,27 +181,30 @@ private void RunWorkers(int degreeOfParallelism, IReadOnlyList<FileInfo> fileLis
 
             var queue = new WorkQueue<FileInfo>();
 
-            /*
-             * Enqueue log files
-             */
-            // TODO: T2.2
+            foreach (var logFile in logFilesToParse)
+            {
+                queue.Enqueue(logFile);
+            }
+            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
+                Thread worker = new Thread(() => WorkerMain(workerId, queue))
+                {
+                    Name = threadName
+                };
+                worker.Start();
+                workers[i] = worker;
             }
 
-            /*
-             * 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 +216,34 @@ 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);
+                    result = new AnalysisResult
+                    (
+                        FileName: file.Name,
+                        FullName: file.FullName,
+                        State: AnalysisState.Succeeded,
+                        Entries: parser.Parse(reader).ToArray(),
+                        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.ToString(),
+                        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;
+                }
             }
         }
     }