feat: concurrency

Author: lideming

Controllers/AdminController.cs

@@ -117,29 +117,41 @@ await _context.Tracks
                     sb.Append("except: ").Append(string.Join(" ", listFail));
                 } else if (arg == "rebuild_loudness") {
                     result = "ok";
-                    var listFail = new List<int>();
-                    var block = new ActionBlock<Track>(async (t) => {
+                    var block = new TransformBlock<Track, (int, TrackAudioInfo)>(async (t) => {
                         try {
                             var info = await _context.App.ConvertService.ComputeLoudnessMap(t);
-                            var oldInfo = await _context.TrackAudioInfos.AsNoTracking().Where(x => x.Id == t.id).FirstOrDefaultAsync();
-                            if (oldInfo == null) {
-                                _context.TrackAudioInfos.Add(info);
-                            } else {
-                                _context.TrackAudioInfos.Update(info);
-                            }
-                            await _context.SaveChangesAsync();
-                        } catch (Exception) {
-                            listFail.Add(t.id);
+                            return (t.id, info);
+                        } catch (Exception e) {
+                            Console.WriteLine(e);
+                            return (t.id, null);
                         }
                     }, new ExecutionDataflowBlockOptions{ MaxDegreeOfParallelism = Environment.ProcessorCount / 2 });
                     await _context.Tracks
                         .Include(t => t.fileRecord)
                         .ForEachAsync((t) => {
                             block.Post(t);
                         });
+                    var listFail = new List<int>();
+                    await foreach (var (id, info) in block.ReceiveAllAsync()) {
+                        if (info == null) {
+                            listFail.Add(id);
+                            continue;
+                        }
+                        try {
+                            var oldInfo = await _context.TrackAudioInfos.AsNoTracking().Where(x => x.Id == id).FirstOrDefaultAsync();
+                            if (oldInfo == null) {
+                                _context.TrackAudioInfos.Add(info);
+                            } else {
+                                _context.TrackAudioInfos.Update(info);
+                            }
+                            await _context.SaveChangesAsync();
+                        } catch (Exception e) {
+                            Console.WriteLine(e);
+                            listFail.Add(id);
+                        }
+                    }
                     block.Complete();
-                    await block.Completion;
-                    sb.Append("failed id list: ").Append(string.Join(" ", listFail));
+                    await _context.SaveChangesAsync();
                 }
             } catch (Exception ex) {
                 result = "error";

Services/ConvertService.cs

@@ -257,9 +257,15 @@ public async Task<TrackAudioInfo> ComputeLoudnessMap(Track track) {
                     } while (haveRead < 4 * 1024);
                     if (haveRead == 0) break;
                     double sum = 0;
-                    for (var i = 0; i < haveRead; i++) {
-                        var x = (sbyte)buffer[i];
-                        sum += x * x;
+                    for (var i = 0; i + 3 < haveRead; i += 4) {
+                        var x0 = (sbyte)buffer[i + 0];
+                        var x1 = (sbyte)buffer[i + 1];
+                        var x2 = (sbyte)buffer[i + 2];
+                        var x3 = (sbyte)buffer[i + 3];
+                        sum += x0 * x0;
+                        sum += x1 * x1;
+                        sum += x2 * x2;
+                        sum += x3 * x3;
                     }
                     var rms = Math.Sqrt(sum / haveRead);
                     ms.WriteByte((byte)rms);