Parallel Parsing (#4)

parse the log lines in parallel

Author: gsquire

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Install SQLite
-        run: sudo apt-get install libsqlite3-dev
+        run: sudo apt-get update && sudo apt-get install libsqlite3-dev
 
       - name: Checkout sources
         uses: actions/checkout@v2

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "topngx"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Garrett Squire <github@garrettsquire.com>"]
 edition = "2018"
 description = "Top for NGINX"
@@ -14,6 +14,7 @@ atty = "0.2"
 env_logger = "0.7"
 log = "0.4"
 once_cell = "1.4"
+rayon = "1.3"
 regex = "1.3"
 rusqlite = "0.23"
 structopt = "0.3"

Cargo.toml

@@ -3,6 +3,7 @@ use std::io::{self, BufRead, BufReader};
 
 use anyhow::{anyhow, Result};
 use log::{debug, info};
+use rayon::prelude::*;
 use regex::Regex;
 use rusqlite::types::ToSql;
 use structopt::StructOpt;
@@ -136,15 +137,16 @@ fn run(opts: &Options, fields: Option<Vec<String>>, queries: Option<Vec<String>>
 }
 
 fn parse_input(input: Box<dyn BufRead>, pattern: &Regex, processor: &Processor) -> Result<()> {
-    let mut records = vec![];
-
-    for line in input.lines() {
-        match pattern.captures(&line?) {
-            None => {}
+    let fields = processor.fields.clone();
+    let lines: Vec<String> = input.lines().filter_map(|l| l.ok()).collect();
+    let records: Vec<_> = lines
+        .par_iter()
+        .filter_map(|line| match pattern.captures(&line) {
+            None => None,
             Some(c) => {
-                let mut record: Vec<(String, Box<dyn ToSql>)> = vec![];
+                let mut record: Vec<(String, Box<dyn ToSql + Send + Sync>)> = vec![];
 
-                for field in &processor.fields {
+                for field in &fields {
                     if field == STATUS_TYPE {
                         let status = c.name("status").map_or("", |m| m.as_str());
                         let status_type = status.parse::<u16>().unwrap_or(0) / 100;
@@ -169,10 +171,10 @@ fn parse_input(input: Box<dyn BufRead>, pattern: &Regex, processor: &Processor)
                     }
                 }
 
-                records.push(record);
+                Some(record)
             }
-        }
-    }
+        })
+        .collect();
 
     processor.process(records)
 }

src/main.rs

@@ -54,7 +54,10 @@ impl Processor {
     }
 
     /// Insert all of the given records into the database.
-    pub(crate) fn process(&self, records: Vec<Vec<(String, Box<dyn ToSql>)>>) -> Result<()> {
+    pub(crate) fn process(
+        &self,
+        records: Vec<Vec<(String, Box<dyn ToSql + Send + Sync>)>>,
+    ) -> Result<()> {
         let insert_stmt = format!(
             "INSERT INTO LOG ({columns}) VALUES ({placeholders})",
             columns = self.columns,