Let FlowData and pipelines release native memory, and close them in the tests and examples

.gitignore

@@ -14,3 +14,7 @@ __pycache__/
 .venv/
 build/
 dist/
+
+# The version each package's setup.py reads, written by the publish and by
+# ci/write-package-versions.ps1.
+version.txt

AGENTS.md

@@ -285,6 +285,15 @@ installed the pipeline packages from PyPI. In CI the root
 `ci/setup-environment.ps1` installs them from `pipeline/` over those,
 so a pipeline change is tested with every product in the same run.
 
+A package's `setup.py` reads its version from a `version.txt`, which only
+`pipeline/fiftyone_pipeline_did` commits, so any other package installed from
+a checkout is 0.0.0. That is below the minimum versions of the pipeline
+packages that the on-premise and examples packages require, so
+`ci/write-package-versions.ps1` gives each published package the version this
+commit would release it at. The root `ci/setup-environment.ps1` runs it
+first. Run `pwsh ci/write-package-versions.ps1` before running those
+packages' tests from a checkout.
+
 ## Submodules
 
 The C/C++ engines are the `cxx` submodule, listed in the root `.gitmodules`.

ci/setup-environment.ps1

@@ -12,6 +12,10 @@ if ($env:GITHUB_JOB -eq "PreBuild") {
     exit 0
 }
 
+# Before anything installs a package from this checkout, so that each one
+# installs at the version this commit would release it at.
+& (Join-Path $PSScriptRoot 'write-package-versions.ps1') -RepoName $RepoName
+
 Invoke-ProductStep 'setup-environment' $RepoName $PSBoundParameters
 
 # The device detection, IP intelligence and location scripts install the

ci/write-package-versions.ps1

@@ -0,0 +1,28 @@
+param (
+    [string]$RepoName = '.'
+)
+$ErrorActionPreference = "Stop"
+. (Join-Path $PSScriptRoot 'packages.ps1')
+
+# Writes the version.txt that each published package's setup.py reads. Without
+# one a package installed from this checkout is 0.0.0, which is below the
+# versions the packages here require of each other, so pip refuses it, or
+# replaces it with a release from PyPI. Each package gets the version this
+# commit would release it at, as the publish gives it when it builds the
+# package, or its last release when this commit would not release it.
+$plan = @(Get-ReleasePlan $RepoName)
+$published = @(Get-Packages $RepoName | Where-Object { $_.Publish -eq $true })
+foreach ($package in $published) {
+    # A version.txt committed with the package is left as it is.
+    if (git -C $RepoName ls-files -- "$($package.Path)/version.txt") {
+        continue
+    }
+    $release = $plan | Where-Object { $_.Name -eq $package.Name }
+    if ($null -ne $release) {
+        $version = $release.Version
+    } else {
+        $version = "$((Get-ReleasedVersions $RepoName $package.Name)[-1])"
+    }
+    $version | Out-File (Join-Path $RepoName $package.Path 'version.txt')
+    Write-Output "$($package.Name) $version"
+}

device-detection/ci/run-integration-tests.ps1

@@ -67,7 +67,12 @@ Push-Location $examplesDir
 try {
     python3 -m venv .virtualenv
     if ($LASTEXITCODE -ne 0) { throw "failed to create the examples virtual environment" }
-    & $py -m pip install -e .
+    # The pipeline packages come from this checkout rather than a package
+    # index, so the examples run with the pipeline code they are released with.
+    $pipeline = 'fiftyone_pipeline_core', 'fiftyone_pipeline_engines',
+        'fiftyone_pipeline_engines_fiftyone', 'fiftyone_pipeline_cloudrequestengine'
+    $local = $pipeline | ForEach-Object { '-e'; "../../pipeline/$_" }
+    & $py -m pip install -e . @local
     if ($LASTEXITCODE -ne 0) { throw "failed to install the examples package" }
 } finally { Pop-Location }
 

device-detection/fiftyone_devicedetection/examples/gettingstarted.py

@@ -35,46 +35,49 @@
 # Cloud
 #pipeline = DeviceDetectionPipelineBuilder({"resource_key": resource_key}).build()
 
-# We create a FlowData object from the pipeline
-# this is used to add evidence to and then process
+# The pipeline is closed at the end of the with block, which releases the
+# native data it holds.
+with pipeline:
+    # We create a FlowData object from the pipeline
+    # this is used to add evidence to and then process. It is closed at the
+    # end of its with block, which releases the results of the detection.
+    with pipeline.create_flowdata() as flowdata1:
 
-flowdata1 = pipeline.create_flowdata()
+        # Here we add a User-Agent of a desktop as evidence
 
-# Here we add a User-Agent of a desktop as evidence
+        desktop_ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.97 Safari/537.36"
 
-desktop_ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.97 Safari/537.36"
+        flowdata1.evidence.add("header.user-agent", desktop_ua)
 
-flowdata1.evidence.add("header.user-agent", desktop_ua)
+        # Now we process the FlowData
 
-# Now we process the FlowData
+        flowdata1.process()
 
-flowdata1.process()
+        # To check whether the User-Agent is a mobile device we look at the ismobile property
+        # inside the Device Detection Engine
 
-# To check whether the User-Agent is a mobile device we look at the ismobile property
-# inside the Device Detection Engine
+        # first we check if this has a meaningful result
 
-# first we check if this has a meaningful result
+        print("Is User-Agent " + desktop_ua + " a mobile? ") 
+        if flowdata1.device.ismobile.has_value():
+            print(flowdata1.device.ismobile.value())
+        else:
+            # Output why the value isn't meaningful
+            print(flowdata1.device.ismobile.no_value_message())
 
-print("Is User-Agent " + desktop_ua + " a mobile? ") 
-if flowdata1.device.ismobile.has_value():
-    print(flowdata1.device.ismobile.value())
-else:
-    # Output why the value isn't meaningful
-    print(flowdata1.device.ismobile.no_value_message())
+    # Now we do the same with a new User-Agent, this time from an iphone
 
-# Now we do the same with a new User-Agent, this time from an iphone
+    with pipeline.create_flowdata() as flowdata2:
 
-flowdata2 = pipeline.create_flowdata()
+        iphone_ua = "Mozilla/5.0 (iPhone; CPU iPhone OS 11_2 like Mac OS X) AppleWebKit/604.4.7 (KHTML, like Gecko) Mobile/15C114"
 
-iphone_ua = "Mozilla/5.0 (iPhone; CPU iPhone OS 11_2 like Mac OS X) AppleWebKit/604.4.7 (KHTML, like Gecko) Mobile/15C114"
+        flowdata2.evidence.add("header.user-agent", iphone_ua)
 
-flowdata2.evidence.add("header.user-agent", iphone_ua)
+        flowdata2.process()
 
-flowdata2.process()
-
-print("Is User-Agent " + iphone_ua + " a mobile device?: ") 
-if flowdata2.device.ismobile.has_value():
-    print(flowdata2.device.ismobile.value())
-else:
-    # Output why the value isn't meaningful
-    print(flowdata2.device.ismobile.no_value_message())
+        print("Is User-Agent " + iphone_ua + " a mobile device?: ") 
+        if flowdata2.device.ismobile.has_value():
+            print(flowdata2.device.ismobile.value())
+        else:
+            # Output why the value isn't meaningful
+            print(flowdata2.device.ismobile.no_value_message())

device-detection/fiftyone_devicedetection_examples/performance-tests/test_performance.py

@@ -48,16 +48,18 @@ def detect(app, pipeline, user_agents):
         # in flowdata.process(). This mock-request doesn't set it, so we set it manually
         request.remote_addr = "127.0.0.1"
 
-        flowdata = pipeline.create_flowdata()
+        # The FlowData is closed when the function returns, which releases
+        # the results of the detection straight away.
+        with pipeline.create_flowdata() as flowdata:
 
-        # Add any information from the request (headers, cookies and additional 
-        # client side provided information)
-        flowdata.evidence.add_from_dict(webevidence(request))
+            # Add any information from the request (headers, cookies and additional 
+            # client side provided information)
+            flowdata.evidence.add_from_dict(webevidence(request))
 
-        # Process the flowdata
-        flowdata.process()
+            # Process the flowdata
+            flowdata.process()
 
-        return getValueHelper(flowdata, "device", "ismobile")
+            return getValueHelper(flowdata, "device", "ismobile")
 
 def benchmark(data_file, user_agents_file, output=False):
     app = Flask(__name__)
@@ -73,13 +75,16 @@ def benchmark(data_file, user_agents_file, output=False):
         settings = { "add_javascript_builder": False },
     ).build()
 
-    with open(user_agents_file) as f:
-        user_agents_list = [ua.rstrip() for ua in f]
-    user_agents = iter(user_agents_list)
+    # The pipeline is closed once the detections have been timed, which
+    # releases the native data it holds.
+    with pipeline:
+        with open(user_agents_file) as f:
+            user_agents_list = [ua.rstrip() for ua in f]
+        user_agents = iter(user_agents_list)
 
-    print(f"Benchmarking with {len(user_agents_list)} user agents", file=sys.stderr)
-    environment = {"detect": detect, "app": app, "pipeline": pipeline, "user_agents": user_agents}
-    time = timeit("detect(app, pipeline, user_agents)", globals=environment, number=len(user_agents_list))
+        print(f"Benchmarking with {len(user_agents_list)} user agents", file=sys.stderr)
+        environment = {"detect": detect, "app": app, "pipeline": pipeline, "user_agents": user_agents}
+        time = timeit("detect(app, pipeline, user_agents)", globals=environment, number=len(user_agents_list))
 
     if output:
         json.dump({

device-detection/fiftyone_devicedetection_examples/setup.py

@@ -65,7 +65,9 @@ def read(file_name):
     # named here so that is visible to anyone reading them.
     # fiftyone_devicedetection is still required because the on-premise
     # examples in this same package use the on-premise engine.
-    install_requires=["fiftyone_devicedetection_cloud", "fiftyone_devicedetection", "flask", "flask-unittest", "json5", "ruamel.yaml"],
+    # fiftyone_pipeline_core 4.5.42 is the first version with close(), which
+    # the examples use.
+    install_requires=["fiftyone_devicedetection_cloud", "fiftyone_devicedetection", "fiftyone_pipeline_core>=4.5.42", "flask", "flask-unittest", "json5", "ruamel.yaml"],
     license="EUPL-1.2",
     classifiers=[
         "Development Status :: 5 - Production/Stable",

device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py

@@ -234,15 +234,18 @@ def run(self, data_file, license_key, interactive, logger, output):
 				auto_update = False,
 				licence_keys = "").add_logger(logger).build()
 
-			# and output the results
-			ExampleUtils.check_data_file(pipeline, logger)
-			if (ExampleUtils.get_data_file_tier(pipeline.get_element("device")) == "Lite"):
-				logger.log("error",
-					"Will not download an 'Enterprise' data file over the top of "
-					"a 'Lite' data file, please supply another location.")
-				raise Exception("File supplied has wrong data tier")
-			logger.log("info", "Existing data file will be replaced with downloaded data file")
-			logger.log("info", f"Existing data file will be copied to {copy_data_file_name}")
+			# The pipeline is closed at the end of the with block, which
+			# releases the native data it holds before the file is replaced.
+			with pipeline:
+				# and output the results
+				ExampleUtils.check_data_file(pipeline, logger)
+				if (ExampleUtils.get_data_file_tier(pipeline.get_element("device")) == "Lite"):
+					logger.log("error",
+						"Will not download an 'Enterprise' data file over the top of "
+						"a 'Lite' data file, please supply another location.")
+					raise Exception("File supplied has wrong data tier")
+				logger.log("info", "Existing data file will be replaced with downloaded data file")
+				logger.log("info", f"Existing data file will be copied to {copy_data_file_name}")
 
 		# do we really want to do this
 		if (interactive):
@@ -295,41 +298,44 @@ def run(self, data_file, license_key, interactive, logger, output):
 			data_update_product_type="V4Enterprise",
 			).add_logger(logger).build()
 
-
-		# thread blocks till update checking is complete - or if there is an
-		# exception we don't get this far
-		update_event.wait()
-		output(f"Update on start-up complete - status - {update_event.status}")
-
-		if update_event.status == UpdateStatus.AUTO_UPDATE_SUCCESS:
-
-			output("Modifying downloaded file to trigger reload - please wait for that"
-				" to complete")
-
-			# wait for the dataUpdateService to notify us that it has updated
-			update_event.clear()
-
-			# it's the same file but changing the file metadata will trigger reload,
-			# demonstrating that if you download a new file and replace the
-			# existing one, then it will be loaded
-			now = datetime.now().timestamp()
-			try:
-				os.utime(data_file, (now, now))
-			except:
-				raise Exception("Could not modify file time, abandoning "
-					"example")
-
-			if update_event.wait(120):
-				output(f"Update on file modification complete, status: {update_event.status}")
-			else:
-				output("Update on file modification timed out")
-		else:
-			logger.log("error", "Auto update was not successful, abandoning example")
-			error_message = f"Auto update failed: {update_event.status}"
-			if update_event.status == UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS:
-				output(error_message)
+		# The pipeline is closed at the end of the with block, which stops
+		# the automatic updates and the file system watcher, and releases
+		# the native data it holds.
+		with pipeline:
+			# thread blocks till update checking is complete - or if there is an
+			# exception we don't get this far
+			update_event.wait()
+			output(f"Update on start-up complete - status - {update_event.status}")
+
+			if update_event.status == UpdateStatus.AUTO_UPDATE_SUCCESS:
+
+				output("Modifying downloaded file to trigger reload - please wait for that"
+					" to complete")
+
+				# wait for the dataUpdateService to notify us that it has updated
+				update_event.clear()
+
+				# it's the same file but changing the file metadata will trigger reload,
+				# demonstrating that if you download a new file and replace the
+				# existing one, then it will be loaded
+				now = datetime.now().timestamp()
+				try:
+					os.utime(data_file, (now, now))
+				except:
+					raise Exception("Could not modify file time, abandoning "
+						"example")
+
+				if update_event.wait(120):
+					output(f"Update on file modification complete, status: {update_event.status}")
+				else:
+					output("Update on file modification timed out")
 			else:
-				raise Exception(error_message)
+				logger.log("error", "Auto update was not successful, abandoning example")
+				error_message = f"Auto update failed: {update_event.status}"
+				if update_event.status == UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS:
+					output(error_message)
+				else:
+					raise Exception(error_message)
 
 		output("Finished Example")