Signal memory pressure relief and expose PSI readings

README.md

@@ -6,7 +6,7 @@ Man pages:
 
 https://man7.org/linux/man-pages/man3/sd_pid_notify.3.html
 https://man7.org/linux/man-pages/man3/sd_listen_fds.3.html
-https://systemd.io/MEMORY_PRESSURE/
+https://systemd.io/PRESSURE/
 
 ## Installation
 
@@ -42,14 +42,6 @@ end
 # Enable systemd watchdog support with `WatchdogSec=5` under `[Service]`
 SystemD.watchdog
 
-# Monitor memory pressure notifications from systemd
-# Enable with `MemoryPressureWatch=auto` and `MemoryPressureThresholdSec=1s` under `[Service]`
-SystemD::MemoryPressure.monitor do
-  # Called when memory pressure is detected
-  # Take action like clearing caches, reducing memory usage, etc.
-  clear_caches
-end
-
 # Store FDs with the SystemD, they will be sent back
 # to the application when it restarts. Requires libsystemd
 clients = Array(TCPSocket).new
@@ -72,6 +64,75 @@ SystemD.named_listeners do |socket, name|
 end
 ```
 
+## Memory pressure
+
+systemd can notify a service when its cgroup is under memory pressure. Enable it under `[Service]`:
+
+```ini
+MemoryPressureWatch=auto
+# Optional, how long tasks may stall on memory within a 2s window before a notification (default 200ms)
+MemoryPressureThresholdSec=200ms
+```
+
+systemd then passes `MEMORY_PRESSURE_WATCH` (the cgroup's `memory.pressure` file) and `MEMORY_PRESSURE_WRITE` (the trigger to register on it) to the service. Without `MEMORY_PRESSURE_WATCH`, or when it's `/dev/null`, monitoring is disabled and the blocks are never called.
+
+### Reacting to pressure
+
+`monitor` calls its block each time memory pressure is detected:
+
+```crystal
+SystemD::MemoryPressure.monitor do
+  # Take action like clearing caches, reducing memory usage, etc.
+  clear_caches
+end
+```
+
+### Pressure and relief
+
+Notifications only signal the onset of pressure, never that it's over. Use `watch` when the application backs off under pressure and must know when to resume. Its block is called with `true` when pressure is detected, and with `false` when it's relieved:
+
+```crystal
+SystemD::MemoryPressure.watch(release_below: 1.0, check_interval: 1.second) do |pressure|
+  if pressure
+    pause_work
+  else
+    resume_work
+  end
+end
+```
+
+While under pressure, `watch` reads the PSI file every `check_interval`. It signals relief once `some avg10` drops below `release_below`, which is the percentage of the last 10 seconds that some task stalled on memory. If no PSI file can be read, relief is signalled at the first check.
+
+`avg10` is a 10-second running average, so it lags behind the stalls. Together with the onset threshold (200ms of stalls in 2s by default), that gives hysteresis: pressure has to be well past before relief is signalled.
+
+Both blocks run in a dedicated thread (an isolated execution context), so keep them short. For example, set a flag and act on it from the application's own fibers.
+
+### Reading PSI values
+
+```crystal
+if pressure = SystemD::MemoryPressure.pressure
+  pressure.some.avg10    # % of time some task stalled on memory, 10s average
+  pressure.some.avg60
+  pressure.some.avg300
+  pressure.some.total    # total stall time in microseconds
+  pressure.full.try &.avg10 # % of time all tasks stalled, nil if there's no "full" line
+end
+```
+
+`pressure` reads the watched file when it's a `.pressure` file, otherwise the process' cgroup v2 `memory.pressure`, falling back to the system-wide `/proc/pressure/memory`. `SystemD::MemoryPressure.parse(string)` parses PSI file contents directly.
+
+### Testing locally
+
+`MEMORY_PRESSURE_WATCH` can point at a regular PSI file, a FIFO or a Unix socket. Writing to a FIFO simulates a notification:
+
+```sh
+mkfifo /tmp/pressure
+MEMORY_PRESSURE_WATCH=/tmp/pressure MEMORY_PRESSURE_WRITE= ./my-app &
+echo > /tmp/pressure
+```
+
+Processes started from a desktop session usually inherit both variables from the desktop's own service, so a shell may already have them set. Clear `MEMORY_PRESSURE_WRITE` when watching a FIFO, as above. Otherwise its trigger is written into the FIFO and read back as a pressure notification.
+
 ## Contributing
 
 1. Fork it (<https://github.com/84codes/systemd.cr/fork>)

spec/memory_pressure_spec.cr

@@ -152,4 +152,121 @@ describe SystemD::MemoryPressure do
       ENV.delete("MEMORY_PRESSURE_WATCH")
     end
   end
+
+  it "parses PSI data" do
+    data = "some avg10=1.50 avg60=0.25 avg300=0.00 total=12345\nfull avg10=0.75 avg60=0.10 avg300=0.00 total=678\n"
+    pressure = SystemD::MemoryPressure.parse(data).should_not be_nil
+    pressure.some.should eq SystemD::MemoryPressure::Stall.new(1.5, 0.25, 0.0, 12345u64)
+    pressure.full.try(&.avg10).should eq 0.75
+  end
+
+  it "parses PSI data without a full line" do
+    pressure = SystemD::MemoryPressure.parse("some avg10=2.00 avg60=1.00 avg300=0.50 total=1\n").should_not be_nil
+    pressure.some.avg10.should eq 2.0
+    pressure.full.should be_nil
+  end
+
+  it "returns nil for malformed PSI data" do
+    SystemD::MemoryPressure.parse("garbage").should be_nil
+    SystemD::MemoryPressure.parse("some avg10=x avg60=1 avg300=1 total=1").should be_nil
+  end
+
+  it "reads pressure from a file" do
+    path = File.tempname
+    File.write(path, "some avg10=3.00 avg60=2.00 avg300=1.00 total=99\n")
+    SystemD::MemoryPressure.pressure(path).try(&.some.total).should eq 99
+    SystemD::MemoryPressure.pressure("/nonexistent/memory.pressure").should be_nil
+  ensure
+    File.delete?(path) if path
+  end
+
+  it "prefers a watched .pressure file" do
+    path = File.tempname(suffix: ".pressure")
+    File.write(path, "some avg10=0.00 avg60=0.00 avg300=0.00 total=0\n")
+    ENV["MEMORY_PRESSURE_WATCH"] = path
+    SystemD::MemoryPressure.pressure_path.should eq path
+  ensure
+    ENV.delete("MEMORY_PRESSURE_WATCH")
+    File.delete?(path) if path
+  end
+
+  it "monitor only calls the block on pressure" do
+    fifo_path = File.tempname
+    begin
+      LibC.mkfifo(fifo_path, 0o600).should eq 0
+      ENV["MEMORY_PRESSURE_WATCH"] = fifo_path
+      ENV.delete("MEMORY_PRESSURE_WRITE")
+
+      calls = Atomic(Int32).new(0)
+      SystemD::MemoryPressure.monitor { calls.add(1) }
+
+      File.open(fifo_path, "w") do |f|
+        f.sync = true
+        f.print "pressure"
+      end
+
+      sleep 100.milliseconds
+      calls.get.should eq 1
+    ensure
+      File.delete(fifo_path) if File.exists?(fifo_path)
+      ENV.delete("MEMORY_PRESSURE_WATCH")
+    end
+  end
+
+  it "watch signals relief once pressure drops below the threshold" do
+    fifo_path = File.tempname
+    begin
+      LibC.mkfifo(fifo_path, 0o600).should eq 0
+      ENV["MEMORY_PRESSURE_WATCH"] = fifo_path
+      ENV.delete("MEMORY_PRESSURE_WRITE")
+
+      pressured = Channel(Nil).new(1)
+      relieved = Channel(Nil).new(1)
+      SystemD::MemoryPressure.watch(Float64::MAX, 10.milliseconds) { |pressure| pressure ? pressured.send(nil) : relieved.send(nil) }
+
+      File.open(fifo_path, "w") do |f|
+        f.sync = true
+        f.print "pressure"
+      end
+
+      pressured.receive
+      select
+      when relieved.receive
+      when timeout(5.seconds)
+        fail "relief was not signalled"
+      end
+    ensure
+      File.delete(fifo_path) if File.exists?(fifo_path)
+      ENV.delete("MEMORY_PRESSURE_WATCH")
+    end
+  end
+
+  it "watch does not signal relief while pressure persists" do
+    pending! "no PSI" unless File.file?("/proc/pressure/memory")
+    fifo_path = File.tempname
+    begin
+      LibC.mkfifo(fifo_path, 0o600).should eq 0
+      ENV["MEMORY_PRESSURE_WATCH"] = fifo_path
+      ENV.delete("MEMORY_PRESSURE_WRITE")
+
+      pressured = Channel(Nil).new(1)
+      relieved = Channel(Nil).new(1)
+      SystemD::MemoryPressure.watch(0.0, 10.milliseconds) { |pressure| pressure ? pressured.send(nil) : relieved.send(nil) }
+
+      File.open(fifo_path, "w") do |f|
+        f.sync = true
+        f.print "pressure"
+      end
+
+      pressured.receive
+      select
+      when relieved.receive
+        fail "relief signalled while under pressure"
+      when timeout(200.milliseconds)
+      end
+    ensure
+      File.delete(fifo_path) if File.exists?(fifo_path)
+      ENV.delete("MEMORY_PRESSURE_WATCH")
+    end
+  end
 end