// ragel -G2 -Z matcher.r6

--- 1   2020-04-27 01:51:27.000000000 +0300
+++ matcher.r6  2020-04-27 01:45:58.000000000 +0300
@@ -0,0 +1,18 @@
+package main
+
+func MatchDate(data []byte) bool {
+    %% machine lexer;
+    %% write data;
+
+    cs, p, pe, eof := 0, 0, len(data), len(data)
+    _ = eof
+
+    %%{
+        main := ^'#' any* '#' ('1'|'5'){2} '111' '0'+ any* @{ return true } ;
+
+        write init;
+        write exec;
+    }%%
+
+    return false
+}

// go run parallelfast.go matcher.go

--- 1   2020-04-27 01:51:27.000000000 +0300
+++ parallelfast.go 2020-04-27 01:46:37.000000000 +0300
@@ -0,0 +1,111 @@
+package main
+
+import (
+   "fmt"
+   "log"
+   "os"
+// "regexp"
+   "sync"
+   "syscall"
+   "time"
+   "unsafe"
+)
+
+const maxSize int64 = 1 << 36
+const numberWorkers = 32
+const lineSize = 65
+
+var waitGroup sync.WaitGroup
+//var regex = regexp.MustCompile("^[^#]*#[15][15]1110+$")
+
+func main() {
+   start := time.Now()
+   log.Printf("Started: %v", start)
+
+   BenchP()
+
+   fmt.Printf("Elapsed time: %v\n", time.Since(start))
+}
+
+func ReadLines(mmap []byte, size int64) [][]byte {
+   var lines [][]byte
+   last_char := int64(0)
+   for i := int64(0); i < size; i++ {
+       if mmap[i] == '\n' {
+           lines = append(lines, mmap[last_char:i])
+           last_char = i + 1
+           i += lineSize
+       }
+   }
+
+   return lines
+}
+
+func QueueLines(queue chan<- [][]byte, lines [][]byte) {
+   linesSz := len(lines)
+   for i := 0; i < linesSz; i += 128 {
+       queue <- lines[i : i+128]
+   }
+}
+
+func BenchP() {
+   ptrSz := int64(unsafe.Sizeof(uint16(0)))
+
+   fd, err := os.Open("log-sample.txt")
+   if err != nil {
+       log.Fatal("Error opening input file:", err)
+   }
+   defer fd.Close()
+
+   fdStat, err := fd.Stat()
+   if err != nil {
+       log.Fatal(err)
+   }
+
+   size := fdStat.Size()
+
+   waitGroup.Add(numberWorkers)
+
+   queue := make(chan [][]byte, size/lineSize)
+   for gr := 1; gr <= numberWorkers; gr++ {
+       go worker(queue, gr, &waitGroup)
+   }
+
+   var mmap []byte
+   mmap, err = syscall.Mmap(int(fd.Fd()), 0, int(size*ptrSz), syscall.PROT_READ, syscall.MAP_SHARED)
+
+   if err = madvise(mmap, syscall.MADV_SEQUENTIAL|syscall.MADV_WILLNEED); err != nil {
+       log.Fatal(err)
+   }
+
+   lines := ReadLines(mmap, size)
+   QueueLines(queue, lines)
+
+   close(queue)
+   waitGroup.Wait()
+}
+
+func madvise(b []byte, advice int) (err error) {
+   _, _, e1 := syscall.Syscall(syscall.SYS_MADVISE, uintptr(unsafe.Pointer(&b[0])), uintptr(len(b)), uintptr(advice))
+   if e1 != 0 {
+       err = e1
+   }
+   return
+}
+
+func worker(queue <-chan [][]byte, id int, wg *sync.WaitGroup) {
+   defer wg.Done()
+   for {
+       lines, ok := <-queue
+       if !ok {
+           fmt.Printf("Worker: %d : Shutting Down\n", id)
+           return
+       }
+       for _, line := range lines {
+//         if regex.Match(line) {
+            if MatchDate(line) {
+               fmt.Printf("[%v] Match: %v\n", id, string(line))
+           }
+       }
+   }
+}