Threading feature (#2)

* Added threaded scanning support.

* Added thread limit checking and thread id.

* Added locking around progress bar updates

* Added support for triggering scanning while fingerprinting

Author: scanoss-cs

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Upcoming changes...
 
+## [0.6.5] - 2021-07-15
+### Added
+- Added support to start scanning while fingerprinting to further increase scan performance
+### Fixed
+- Ignoring broken symlink files
+
 ## [0.6.0] - 2021-07-14
 ### Added
 - Added threading to speed up fingerprint scanning
@@ -34,3 +40,4 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 [0.5.5]: https://github.com/scanoss/scanoss.py/compare/v0.5.4...v0.5.5
 [0.5.6]: https://github.com/scanoss/scanoss.py/compare/v0.5.5...v0.5.6
 [0.6.0]: https://github.com/scanoss/scanoss.py/compare/v0.5.6...v0.6.0
+[0.6.5]: https://github.com/scanoss/scanoss.py/compare/v0.6.0...v0.6.5

src/scanoss/__init__.py

@@ -17,4 +17,4 @@
    along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 
-__version__ = '0.6.0'
+__version__ = '0.6.5'

src/scanoss/scanner.py

@@ -254,122 +254,91 @@ def scan_folder(self, scan_dir: str) -> bool:
             raise Exception(f"ERROR: Please specify a folder to scan")
         if not os.path.exists(scan_dir) or not os.path.isdir(scan_dir):
             raise Exception(f"ERROR: Specified folder does not exist or is not a folder: {scan_dir}")
-        wfps = ''
+
         scan_dir_len = len(scan_dir) if scan_dir.endswith(os.path.sep) else len(scan_dir)+1
         self.print_msg(f'Searching {scan_dir} for files to fingerprint...')
         spinner = None
         if not self.quiet and self.isatty:
             spinner = Spinner('Fingerprinting ')
+        wfp_list = []
+        scan_block = ''
+        scan_size = 0
+        queue_size = 0
+        file_count = 0
+        scan_started = False
         for root, dirs, files in os.walk(scan_dir):
-            self.print_debug(f'U Root: {root}, Dirs: {dirs}, Files {files}')
-            dirs[:] = Scanner.__filter_dirs(dirs)                          # Strip out unwanted directories
-            filtered_files = Scanner.__filter_files(files)                 # Strip out unwanted files
+            self.print_trace(f'U Root: {root}, Dirs: {dirs}, Files {files}')
+            dirs[:] = Scanner.__filter_dirs(dirs)                             # Strip out unwanted directories
+            filtered_files = Scanner.__filter_files(files)                    # Strip out unwanted files
             self.print_debug(f'F Root: {root}, Dirs: {dirs}, Files {filtered_files}')
-            for file in filtered_files:
+            for file in filtered_files:                                       # Cycle through each filtered file
                 path = os.path.join(root, file)
-                file_stat = os.stat(path)
-                if file_stat.st_size > 0:            # Ignore empty files
-                    self.print_debug(f'Fingerprinting {path}...')
+                f_size = 0
+                try:
+                    f_size = os.stat(path).st_size
+                except:
+                    self.print_trace(f'Ignoring missing symlink file: {file}') # Can fail if there is a broken symlink
+                if f_size > 0:                                                 # Ignore broken links and empty files
+                    self.print_trace(f'Fingerprinting {path}...')
                     if spinner:
                         spinner.next()
-                    wfps += self.winnowing.wfp_for_file(path, Scanner.__strip_dir(scan_dir, scan_dir_len, path))
+                    wfp = self.winnowing.wfp_for_file(path, Scanner.__strip_dir(scan_dir, scan_dir_len, path))
+                    wfp_list.append(wfp)
+                    file_count += 1
+                    if self.threaded_scan:
+                        wfp_size = len(wfp.encode("utf-8"))
+                        if (wfp_size + scan_size) >= MAX_POST_SIZE:
+                            self.threaded_scan.queue_add(scan_block)
+                            queue_size += 1
+                            scan_block = ''
+                        scan_block += wfp
+                        scan_size = len(scan_block.encode("utf-8"))
+                        if scan_size >= MAX_POST_SIZE:
+                            self.threaded_scan.queue_add(scan_block)
+                            queue_size += 1
+                            scan_block = ''
+                        if queue_size > self.nb_threads and not scan_started: # Start scanning if we have something to do
+                            scan_started = True
+                            if not self.threaded_scan.run(wait=False):
+                                self.print_stderr(
+                                    f'Warning: Some errors encounted while scanning. Results might be incomplete.')
+                                success = False
+        # End for loop
+        if self.threaded_scan and scan_block:
+            self.threaded_scan.queue_add(scan_block)  # Make sure all files have been submitted
         if spinner:
             spinner.finish()
-        if wfps:
+
+        if wfp_list:
             self.print_debug(f'Writing fingerprints to {self.wfp}')
             with open(self.wfp, 'w') as f:
-                f.write(wfps)
-            self.print_msg(f'Scanning fingerprints...')
+                f.write(''.join(wfp_list))
             if self.scan_output:
                 self.print_msg(f'Writing results to {self.scan_output}...')
             if self.threaded_scan:
-                success = self.scan_wfp_file_threaded()
+                success = self.__finish_scan_threaded(scan_started, file_count)
             else:
                 success = self.scan_wfp_file()
         else:
             Scanner.print_stderr(f'Warning: No files found to scan in folder: {scan_dir}')
         return success
 
-    def scan_file(self, file: str) -> bool:
+    def __finish_scan_threaded(self, scan_started: bool, file_count: int) -> bool:
         """
-        Scan the specified file and produce a result
-        Parameters
-        ----------
-            file: str
-                File to fingerprint and scan/identify
-        :return True if successful, False otherwise
+        Finish scanning the filtered files and wait for the threads to complete
+        :param scan_started: If the scan has already started or not
+        :param file_count:  Number of total files to be scanned
+        :return: True if successful, False otherwise
         """
         success = True
-        if not file:
-            raise Exception(f"ERROR: Please specify a file to scan")
-        if not os.path.exists(file) or not os.path.isfile(file):
-            raise Exception(f"ERROR: Specified files does not exist or is not a file: {file}")
-        self.print_debug(f'Fingerprinting {file}...')
-        wfps = self.winnowing.wfp_for_file(file, file)
-        if wfps:
-            self.print_debug(f'Scanning {file}...')
-            if self.scan_output:
-                self.print_msg(f'Writing results to {self.scan_output}...')
-            success = self.scan_wfp(wfps)
+        self.threaded_scan.update_bar(create=True, file_count=file_count)
+        if not scan_started:
+            if not self.threaded_scan.run():           # Run the scan and wait for it to complete
+                self.print_stderr(f'Warning: Some errors encounted while scanning. Results might be incomplete.')
+                success = False
         else:
-            success = False
-
-        return success
-
-    def scan_wfp_file_threaded(self, file: str = None) -> bool:
-        """
-        Scan the supplied WFP file in multiple threads
-        :param file: WFP file to scan
-        :return True if scuccessful, False otherwise
-        """
-        success = True
-        wfp_file = file if file else self.wfp   # If a WFP file is specified, use it, otherwise us the default
-        if not os.path.exists(wfp_file) or not os.path.isfile(wfp_file):
-            raise Exception(f"ERROR: Specified WFP file does not exist or is not a file: {wfp_file}")
-        file_count = Scanner.__count_files_in_wfp_file(wfp_file)
-        cur_files = 0
-        cur_size = 0
-        batch_files = 0
-        wfp = ''
-        self.print_debug(f'Found {file_count} files to process.')
-        file_print = ''
-        bar = None
-        if not self.quiet and self.isatty:
-            bar = Bar('Scanning', max=file_count)
-            bar.next(0)
-            self.threaded_scan.set_bar(bar)
-
-        with open(wfp_file) as f:
-            for line in f:
-                if line.startswith(WFP_FILE_START):
-                    if file_print:
-                        wfp += file_print         # Store the WFP for the current file
-                        cur_size = len(wfp.encode("utf-8"))
-                    file_print = line             # Start storing the next file
-                    cur_files += 1
-                    batch_files += 1
-                else:
-                    file_print += line             # Store the rest of the WFP for this file
-                l_size = cur_size + len(file_print.encode('utf-8'))
-                # Hit the max post size, so sending the current batch and continue processing
-                if l_size >= MAX_POST_SIZE and wfp:
-                    self.print_debug(f'Added {batch_files} ({cur_files}) of'
-                                     f' {file_count} ({len(wfp.encode("utf-8"))} bytes) files to the pending queue.')
-                    if cur_size > MAX_POST_SIZE:
-                        Scanner.print_stderr(f'Warning: Post size {cur_size} greater than limit {MAX_POST_SIZE}')
-                    self.threaded_scan.queue_add(wfp)
-                    batch_files = 0
-                    wfp = ''
-        if file_print:
-            wfp += file_print  # Store the WFP for the current file
-        if wfp:
-            self.print_debug(f'Adding {batch_files} ({cur_files}) of'
-                             f' {file_count} ({len(wfp.encode("utf-8"))} bytes) files to the pending queue.')
-            self.threaded_scan.queue_add(wfp)
-
-        if not self.threaded_scan.run():
-            self.print_stderr(f'Warning: Some errors encounted while scanning. Result might not be complete.')
-            success = False
+            self.threaded_scan.complete()               # Wait for the scans to complete
+        self.threaded_scan.complete_bar()
         responses = self.threaded_scan.responses
         raw_output = "{\n"
         if responses:
@@ -382,11 +351,10 @@ def scan_wfp_file_threaded(self, file: str = None) -> bool:
                             first = False
                         else:
                             raw_output += ",\n  \"%s\":%s" % (key, json.dumps(value, indent=2))
+            # End for loop
         else:
             success = False
         raw_output += "\n}"
-        if bar:
-            bar.finish()
         parsed_json = None
         try:
             parsed_json = json.loads(raw_output)
@@ -404,6 +372,32 @@ def scan_wfp_file_threaded(self, file: str = None) -> bool:
                 success = cdx.produce_from_json(parsed_json)
             else:
                 success = cdx.produce_from_str(raw_output)
+        return success
+
+
+    def scan_file(self, file: str) -> bool:
+        """
+        Scan the specified file and produce a result
+        Parameters
+        ----------
+            file: str
+                File to fingerprint and scan/identify
+        :return True if successful, False otherwise
+        """
+        success = True
+        if not file:
+            raise Exception(f"ERROR: Please specify a file to scan")
+        if not os.path.exists(file) or not os.path.isfile(file):
+            raise Exception(f"ERROR: Specified files does not exist or is not a file: {file}")
+        self.print_debug(f'Fingerprinting {file}...')
+        wfp = self.winnowing.wfp_for_file(file, file)
+        if wfp:
+            self.print_debug(f'Scanning {file}...')
+            if self.scan_output:
+                self.print_msg(f'Writing results to {self.scan_output}...')
+            success = self.scan_wfp(wfp)
+        else:
+            success = False
 
         return success
 
@@ -538,14 +532,14 @@ def wfp_file(self, scan_file: str, wfp_file: str = None):
             raise Exception(f"ERROR: Specified file does not exist or is not a file: {scan_file}")
 
         self.print_debug(f'Fingerprinting {scan_file}...')
-        wfps = self.winnowing.wfp_for_file(scan_file, scan_file)
-        if wfps:
+        wfp = self.winnowing.wfp_for_file(scan_file, scan_file)
+        if wfp:
             if wfp_file:
                 self.print_stderr(f'Writing fingerprints to {wfp_file}')
                 with open(wfp_file, 'w') as f:
-                    f.write(wfps)
+                    f.write(wfp)
             else:
-                print(wfps)
+                print(wfp)
         else:
             Scanner.print_stderr(f'Warning: No fingerprints generated for: {scan_file}')
 
@@ -561,7 +555,7 @@ def wfp_folder(self, scan_dir: str, wfp_file: str = None):
         scan_dir_len = len(scan_dir) if scan_dir.endswith(os.path.sep) else len(scan_dir)+1
         self.print_msg(f'Searching {scan_dir} for files to fingerprint...')
         for root, dirs, files in os.walk(scan_dir):
-            dirs = Scanner.__filter_dirs(dirs)                             # Strip out unwanted directories
+            dirs[:] = Scanner.__filter_dirs(dirs)                             # Strip out unwanted directories
             filtered_files = Scanner.__filter_files(files)                    # Strip out unwanted files
             self.print_trace(f'Root: {root}, Dirs: {dirs}, Files {filtered_files}')
             for file in filtered_files:

src/scanoss/threadedscanning.py

@@ -25,7 +25,6 @@
 from typing import Dict, List
 from dataclasses import dataclass
 from progress.bar import Bar
-from progress.spinner import Spinner
 
 from .scanossapi import ScanossApi
 
@@ -41,6 +40,7 @@ class ThreadedScanning(object):
     """
     inputs: queue.Queue = queue.Queue()
     output: queue.Queue = queue.Queue()
+    bar: Bar = None
 
     def __init__(self, scanapi :ScanossApi, debug: bool = False, trace: bool = False, quiet: bool = False,
                  nb_threads: int = 5
@@ -57,9 +57,11 @@ def __init__(self, scanapi :ScanossApi, debug: bool = False, trace: bool = False
         self.debug = debug
         self.trace = trace
         self.quiet = quiet
+        self.isatty = sys.stderr.isatty()
         self.nb_threads = nb_threads
-        self.bar = None
+        self.bar_count = 0
         self.errors = False
+        self.lock = threading.Lock()
         if nb_threads > MAX_ALLOWED_THREADS:
             self.print_msg(f'Warning: Requested threads too large: {nb_threads}. Reducing to {MAX_ALLOWED_THREADS}')
             self.nb_threads = MAX_ALLOWED_THREADS
@@ -106,20 +108,39 @@ def print_trace(self, *args, **kwargs):
         if self.trace:
             self.print_stderr(*args, **kwargs)
 
+    def create_bar(self, file_count: int):
+        if not self.quiet and self.isatty and not self.bar:
+            self.bar = Bar('Scanning', max=file_count)
+            self.bar.next(self.bar_count)
+
+    def complete_bar(self):
+        if self.bar:
+            self.bar.finish()
+
     def set_bar(self, bar: Bar) -> None:
         """
         Set the Progress Bar to display progress while scanning
         :param bar: Progress Bar object
         """
         self.bar = bar
 
-    def update_bar(self, amount: int) -> None:
+    def update_bar(self, amount: int = 0, create: bool = False, file_count: int = 0) -> None:
         """
         Update the Progress Bar progress
         :param amount: amount of progress to update
         """
-        if self.bar:
-            self.bar.next(amount)
+        try:
+            self.lock.acquire()
+            try:
+                if create and not self.bar:
+                    self.create_bar(file_count)
+                elif self.bar:
+                    self.bar.next(amount)
+                self.bar_count += amount
+            finally:
+                self.lock.release()
+        except Exception as e:
+            self.print_debug(f'Warning: Update status bar lock failed: {e}. Ignoring.')
 
     def queue_add(self, wfp: str) -> None:
         """
@@ -128,6 +149,9 @@ def queue_add(self, wfp: str) -> None:
         """
         self.inputs.put(wfp)
 
+    def get_queue_size(self) -> int:
+        return self.inputs.qsize()
+
     @property
     def responses(self) -> List[Dict]:
         """
@@ -136,7 +160,7 @@ def responses(self) -> List[Dict]:
         """
         return list(self.output.queue)
 
-    def run(self) -> bool:
+    def run(self, wait: bool = True) -> bool:
         """
         Initiate the threads and process all pending requests
         :return: True if successful, False if error encountered
@@ -154,10 +178,16 @@ def run(self) -> bool:
         except Exception as e:
             self.print_stderr(f'ERROR: Problem running threaded scanning: {e}')
             self.errors = True
-        self.inputs.join()
-
+        if wait:                    # Wait for all inputs to complete
+            self.inputs.join()
         return False if self.errors else True
 
+    def complete(self) -> None:
+        """
+        Wait for input queue to complete processing
+        """
+        self.inputs.join()
+
     def worker_post(self) -> None:
         """
         Take each request and process it