File Uploads, Downloads, Multipart Requests, and Large Payloads: Guided Hands-On Workflow

1. Safety envelope

2. Generate bounded payloads and manifest before timing

Save as tools/generate_payloads.py and run from the lab root:

from pathlib import Path
import csv
import hashlib
import json

SPECS = [
    ("small", 64 * 1024),
    ("medium", 512 * 1024),
]
ROOT = Path("data/payloads").resolve()
ROOT.mkdir(parents=True, exist_ok=True)

def write_payload(name, size_bytes):
    path = ROOT / f"payload-{name}.txt"
    seed = (
        f"PROMPT18|{name}|synthetic-only|"
        "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz|\n"
    ).encode("ascii")
    sha256 = hashlib.sha256()
    md5 = hashlib.md5()
    remaining = size_bytes
    with path.open("wb") as handle:
        while remaining:
            chunk = seed[:min(len(seed), remaining)]
            handle.write(chunk)
            sha256.update(chunk)
            md5.update(chunk)
            remaining -= len(chunk)
    return {
        "payload_name": name,
        "payload_path": path.relative_to(Path.cwd().resolve()).as_posix(),
        "expected_bytes": size_bytes,
        "expected_sha256": sha256.hexdigest(),
        "expected_md5": md5.hexdigest(),
        "mime_type": "text/plain",
    }

rows = [write_payload(name, size) for name, size in SPECS]
Path("data").mkdir(exist_ok=True)

Path("data/payload-manifest.json").write_text(
    json.dumps({"version": 1, "payloads": rows}, indent=2),
    encoding="utf-8",
)

with Path("data/payloads.csv").open("w", newline="", encoding="utf-8") as handle:
    writer = csv.DictWriter(handle, fieldnames=rows[0].keys())
    writer.writeheader()
    writer.writerows(rows)

for row in rows:
    print(
        f"{row['payload_name']}: {row['expected_bytes']} bytes "
        f"sha256={row['expected_sha256']} md5={row['expected_md5']} "
        f"path={row['payload_path']}"
    )

python tools/generate_payloads.py

This creates:

• data/payloads/payload-small.txt — exactly 65,536 bytes;
• data/payloads/payload-medium.txt — exactly 524,288 bytes;
• data/payload-manifest.json;
• data/payloads.csv for JMeter.

The generator is deliberately outside JMeter's timed HTTP samplers. Its CPU/disk work therefore does not masquerade as server upload latency.

3. Start the guarded localhost service

Save as fixtures/upload_download_fixture.py:

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from email.parser import BytesParser
from email.policy import default
from pathlib import Path
from urllib.parse import urlparse
import argparse
import hashlib
import itertools
import json
import re
import threading
import time

FIXTURE_VERSION = "prompt18-upload-download-fixture-v1"
MAX_FILE_BYTES = 1024 * 1024
MAX_REQUEST_BYTES = 2 * 1024 * 1024
SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,64}$")

lock = threading.Lock()
upload_seq = itertools.count(1)
registry = {}
event_log = None
storage_root = None
metrics = {
    "requests": 0,
    "errors": 0,
    "uploads": 0,
    "downloads": 0,
    "deletes": 0,
    "request_body_bytes": 0,
    "download_payload_bytes": 0,
    "peak_storage_bytes": 0,
}

def now_ms():
    return int(time.time() * 1000)

def storage_bytes():
    return sum(int(item["bytes"]) for item in registry.values())

def snapshot_storage():
    current = storage_bytes()
    metrics["peak_storage_bytes"] = max(metrics["peak_storage_bytes"], current)
    return current

def log_event(event):
    if event_log is None:
        return
    with lock:
        with event_log.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, sort_keys=True) + "\n")

def parse_multipart(content_type, body):
    prefix = (
        f"Content-Type: {content_type}\r\n"
        "MIME-Version: 1.0\r\n\r\n"
    ).encode("utf-8")
    message = BytesParser(policy=default).parsebytes(prefix + body)
    if not message.is_multipart():
        raise ValueError("not multipart")
    fields = {}
    file_part = None
    for part in message.iter_parts():
        name = part.get_param("name", header="content-disposition")
        filename = part.get_filename()
        payload = part.get_payload(decode=True) or b""
        if filename is not None:
            if name == "file":
                file_part = {
                    "filename": filename,
                    "content_type": part.get_content_type(),
                    "payload": payload,
                }
        elif name:
            fields[name] = payload.decode("utf-8", errors="strict")
    if file_part is None:
        raise ValueError("missing file field")
    return fields, file_part

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def send_json(self, status, payload):
        raw = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        self.end_headers()
        self.wfile.write(raw)

    def send_empty(self, status):
        self.send_response(status)
        self.send_header("Content-Length", "0")
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        self.end_headers()

    def record(self, started, operation, status, **extra):
        with lock:
            metrics["requests"] += 1
            if status >= 400:
                metrics["errors"] += 1
            current_storage = snapshot_storage()
        event = {
            "ts_ms": now_ms(),
            "operation": operation,
            "method": self.command,
            "path": urlparse(self.path).path,
            "status": status,
            "service_wall_ms": now_ms() - started,
            "storage_bytes_after": current_storage,
        }
        event.update(extra)
        log_event(event)

    def do_GET(self):
        started = now_ms()
        path = urlparse(self.path).path

        if path == "/health":
            self.send_json(200, {
                "status": "ok",
                "fixture_version": FIXTURE_VERSION,
                "max_file_bytes": MAX_FILE_BYTES,
                "max_request_bytes": MAX_REQUEST_BYTES,
            })
            self.record(started, "health", 200)
            return

        if path == "/stats":
            with lock:
                items = [
                    {
                        "upload_id": upload_id,
                        "run_id": item["run_id"],
                        "payload_name": item["payload_name"],
                        "bytes": item["bytes"],
                        "sha256": item["sha256"],
                        "md5": item["md5"],
                    }
                    for upload_id, item in sorted(registry.items())
                ]
                current_storage = storage_bytes()
                snap_metrics = dict(metrics)
            self.send_json(200, {
                "fixture_version": FIXTURE_VERSION,
                "stored_uploads": len(items),
                "stored_bytes": current_storage,
                "items": items,
                "metrics": snap_metrics,
            })
            self.record(started, "stats", 200)
            return

        if path.startswith("/download/"):
            upload_id = path.split("/", 2)[2]
            with lock:
                item = registry.get(upload_id)
                item_copy = dict(item) if item else None
            if item_copy is None:
                self.send_json(404, {"status": "not_found", "upload_id": upload_id})
                self.record(started, "download", 404, upload_id=upload_id)
                return

            file_path = Path(item_copy["path"])
            if not file_path.is_file():
                self.send_json(410, {"status": "storage_missing", "upload_id": upload_id})
                self.record(started, "download", 410, upload_id=upload_id)
                return

            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=us-ascii")
            self.send_header("Content-Length", str(item_copy["bytes"]))
            self.send_header("X-Upload-ID", upload_id)
            self.send_header("X-Payload-SHA256", item_copy["sha256"])
            self.send_header("X-Payload-MD5", item_copy["md5"])
            self.send_header("X-Fixture-Version", FIXTURE_VERSION)
            self.end_headers()

            with file_path.open("rb") as handle:
                while True:
                    chunk = handle.read(64 * 1024)
                    if not chunk:
                        break
                    self.wfile.write(chunk)
            self.wfile.flush()

            with lock:
                metrics["downloads"] += 1
                metrics["download_payload_bytes"] += int(item_copy["bytes"])
            self.record(
                started, "download", 200,
                upload_id=upload_id,
                run_id=item_copy["run_id"],
                payload_name=item_copy["payload_name"],
                payload_bytes=item_copy["bytes"],
                sha256=item_copy["sha256"],
                md5=item_copy["md5"],
            )
            return

        self.send_json(404, {"status": "not_found", "path": path})
        self.record(started, "unknown_get", 404)

    def do_POST(self):
        started = now_ms()
        path = urlparse(self.path).path

        if path != "/upload":
            self.send_json(404, {"status": "not_found", "path": path})
            self.record(started, "unknown_post", 404)
            return

        content_type = self.headers.get("Content-Type", "")
        try:
            content_length = int(self.headers.get("Content-Length", "0") or "0")
        except ValueError:
            content_length = -1

        if content_length < 0 or content_length > MAX_REQUEST_BYTES:
            self.send_json(413, {"status": "request_too_large", "max_request_bytes": MAX_REQUEST_BYTES})
            self.record(started, "upload", 413, request_content_length=content_length)
            return

        body = self.rfile.read(content_length)
        with lock:
            metrics["request_body_bytes"] += len(body)

        if "multipart/form-data" not in content_type.lower():
            self.send_json(415, {"status": "multipart_required"})
            self.record(started, "upload", 415, request_content_length=len(body))
            return

        try:
            fields, file_part = parse_multipart(content_type, body)
        except Exception:
            self.send_json(400, {"status": "invalid_multipart"})
            self.record(started, "upload", 400, request_content_length=len(body))
            return

        run_id = fields.get("run_id", "")
        payload_name = fields.get("payload_name", "")
        if not SAFE_TOKEN.fullmatch(run_id) or not SAFE_TOKEN.fullmatch(payload_name):
            self.send_json(400, {"status": "invalid_metadata"})
            self.record(started, "upload", 400, request_content_length=len(body))
            return

        payload = file_part["payload"]
        if len(payload) > MAX_FILE_BYTES:
            self.send_json(413, {"status": "file_too_large", "max_file_bytes": MAX_FILE_BYTES})
            self.record(started, "upload", 413, request_content_length=len(body))
            return

        sha256 = hashlib.sha256(payload).hexdigest()
        md5 = hashlib.md5(payload).hexdigest()

        with lock:
            upload_id = f"U{next(upload_seq):06d}"
            target = (storage_root / f"{upload_id}.txt").resolve()
            if target.parent != storage_root:
                raise RuntimeError("storage guard failed")
            target.write_bytes(payload)
            registry[upload_id] = {
                "run_id": run_id,
                "payload_name": payload_name,
                "bytes": len(payload),
                "sha256": sha256,
                "md5": md5,
                "path": str(target),
            }
            metrics["uploads"] += 1
            current_storage = snapshot_storage()

        response = {
            "status": "stored",
            "upload_id": upload_id,
            "run_id": run_id,
            "payload_name": payload_name,
            "bytes": len(payload),
            "sha256": sha256,
            "md5": md5,
            "request_content_length": len(body),
            "storage_bytes_after": current_storage,
        }
        self.send_json(201, response)
        self.record(
            started, "upload", 201,
            upload_id=upload_id,
            run_id=run_id,
            payload_name=payload_name,
            payload_bytes=len(payload),
            request_content_length=len(body),
            sha256=sha256,
            md5=md5,
        )

    def do_DELETE(self):
        started = now_ms()
        path = urlparse(self.path).path
        if not path.startswith("/upload/"):
            self.send_json(404, {"status": "not_found", "path": path})
            self.record(started, "unknown_delete", 404)
            return

        upload_id = path.split("/", 2)[2]
        with lock:
            item = registry.pop(upload_id, None)
            if item:
                file_path = Path(item["path"])
                if file_path.is_file() and file_path.parent.resolve() == storage_root:
                    file_path.unlink()
                metrics["deletes"] += 1

        if item is None:
            self.send_json(404, {"status": "not_found", "upload_id": upload_id})
            self.record(started, "delete", 404, upload_id=upload_id)
            return

        self.send_empty(204)
        self.record(
            started, "delete", 204,
            upload_id=upload_id,
            run_id=item["run_id"],
            payload_name=item["payload_name"],
            payload_bytes=item["bytes"],
        )

    def log_message(self, format, *args):
        return

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8018)
    parser.add_argument("--root", default="lab-storage")
    parser.add_argument("--log", default="results/server-events.jsonl")
    args = parser.parse_args()

    root = Path(args.root).resolve()
    root.mkdir(parents=True, exist_ok=True)

    global storage_root, event_log
    storage_root = root
    event_log = Path(args.log).resolve()
    event_log.parent.mkdir(parents=True, exist_ok=True)
    event_log.write_text("", encoding="utf-8")

    print(f"fixture_version={FIXTURE_VERSION}")
    print(f"listen=http://{args.host}:{args.port}")
    print(f"storage_root={storage_root}")
    print(f"event_log={event_log}")

    server = ThreadingHTTPServer((args.host, args.port), Handler)
    server.serve_forever()

if __name__ == "__main__":
    main()

PowerShell:

New-Item -ItemType Directory -Force .\results | Out-Null
python .\fixtures\upload_download_fixture.py `
  --host 127.0.0.1 `
  --port 8018 `
  --root .\lab-storage `
  --log .\results\server-events.jsonl

Bash:

mkdir -p results
python fixtures/upload_download_fixture.py   --host 127.0.0.1   --port 8018   --root ./lab-storage   --log ./results/server-events.jsonl

The fixture never trusts a client-supplied filesystem path. It assigns U000001... IDs and stores only under the resolved guarded root.

4. Read-only preflight

curl --fail --silent http://127.0.0.1:8018/health
curl --fail --silent http://127.0.0.1:8018/stats

Require fixture version prompt18-upload-download-fixture-v1, stored_uploads=0, stored_bytes=0, max file=1 MiB. Also inspect the manifest and free disk before traffic.

5. Record generator disk/result baseline

Save tools/disk_snapshot.py:

from pathlib import Path
import shutil
import sys

root = Path(sys.argv[1] if len(sys.argv) > 1 else ".").resolve()
usage = shutil.disk_usage(root)
print(f"root={root}")
print(f"disk_total={usage.total}")
print(f"disk_used={usage.used}")
print(f"disk_free={usage.free}")

for raw in sys.argv[2:]:
    p = Path(raw)
    if p.exists():
        if p.is_file():
            print(f"file_bytes {p}={p.stat().st_size}")
        elif p.is_dir():
            total = sum(f.stat().st_size for f in p.rglob("*") if f.is_file())
            print(f"directory_bytes {p}={total}")

python tools/disk_snapshot.py . data results lab-storage

Record these values before and after each profile. This does not measure JVM heap, but it makes payload/result/server-storage disk growth explicit. With a Java 17 JDK, jcmd can optionally sample a long-enough JMeter process (jcmd to list JVMs, then jcmd <pid> GC.heap_info) without changing the plan.

6. HTTP Request Defaults and run identity

HTTP Request Defaults:

• Protocol: http
• Server: 127.0.0.1
• Port: 8018
• Connect timeout: 500 ms
• Response timeout: 3000 ms

Define RUN_ID=${__P(RUN_ID,prompt18-local)} as an immutable run-level property/variable. It is safe as a property because it identifies the whole run; per-upload UPLOAD_ID remains thread-local.

7. CSV Data Set Config

Use data/payloads.csv:

Variables become payload_name, payload_path, expected_bytes, expected_sha256, expected_md5, and mime_type.

8. Multipart upload sampler

HTTP Request — Upload Payload:

• POST /upload
• Parameters: run_id=${RUN_ID}, payload_name=${payload_name}
• File Path: ${payload_path}
• Parameter name: file
• MIME Type: ${mime_type}
• Use multipart/form-data: enabled/automatic because a file path is supplied
• Browser-compatible headers: leave unchecked for this fixture; the server accepts the normal per-part headers.

Run JMeter from the lab root so relative file paths resolve consistently. Do not put a per-thread generated file writer beside this timed sampler.

9. Extract and validate upload metadata

Add JSON JMESPath Extractors:

Assert HTTP 201/status=stored, SERVER_BYTES=${expected_bytes}, SERVER_SHA256=${expected_sha256}, and SERVER_MD5=${expected_md5}. This proves the server stored the same payload content without inspecting a huge response body.

10. Lean download with MD5 retention

HTTP Request — Download Payload:

• GET /download/${UPLOAD_ID}
• Save response as MD5 hash = checked

Add Response Assertion on Response Text/Body equals ${expected_md5} (the SampleResult body is the 32-character MD5 in this mode). Also assert HTTP 200.

The network still transfers the full 64/512 KiB response. JMeter simply avoids retaining those original bytes in the SampleResult.

11. Delete exactly what this thread uploaded

HTTP Request — Delete Upload:

• DELETE /upload/${UPLOAD_ID}
• assert HTTP 204.

The server only deletes IDs present in its in-memory registry and confirms that the file's parent is exactly the guarded storage root.

12. Progressive tree

Test Plan
├── User Defined Variables -> RUN_ID
├── HTTP Request Defaults -> 127.0.0.1:8018
├── CSV Data Set Config -> data/payloads.csv
└── Thread Group
    ├── Upload Payload (multipart)
    │   ├── JMESPath extract UPLOAD_ID / SERVER_BYTES / SERVER_SHA256 / SERVER_MD5
    │   └── assertions against manifest variables
    ├── Download Payload
    │   ├── Save response as MD5 hash = true
    │   └── assert response body == ${expected_md5}
    └── Delete Upload
        └── assert 204

13. One-thread sequential size comparison first

Run 1 thread ×2 loops in GUI with View Results Tree only for this authoring pass. Because CSV has two rows, one thread observes small then medium. Confirm:

• upload server bytes exactly 65,536 then 524,288;
• upload JTL sentBytes is larger than payload bytes because multipart/HTTP adds overhead;
• download response shown to JMeter is a 32-char MD5 rather than the original body;
• server stored bytes return to zero after each delete.

Disable View Results Tree before meaningful runs.

14. Lean CLI run

PowerShell:

jmeter.bat -n `
  -t plans\transfer-lean.jmx `
  -l results\lean\results.jtl `
  -j results\lean\jmeter.log `
  -JRUN_ID=prompt18-lean `
  -Jsample_variables=payload_name,expected_bytes

The default CSV JTL retains byte counters without response bodies. Keep jmeter.save.saveservice.response_data=false, request/response headers false, and samplerData false.

15. Analyze JTL byte/timing evidence

Save tools/analyze_transfer_jtl.py:

import csv
import math
import statistics
import sys
from collections import defaultdict
from pathlib import Path

path = Path(sys.argv[1])
rows = list(csv.DictReader(path.open(newline="", encoding="utf-8")))
if not rows:
    raise SystemExit("No JTL rows")

required = {"elapsed", "label", "success", "responseCode", "bytes", "sentBytes"}
missing = required - set(rows[0])
if missing:
    raise SystemExit(f"Missing JTL fields: {sorted(missing)}")

def percentile(values, pct):
    data = sorted(values)
    if not data:
        return 0
    idx = max(0, min(len(data) - 1, math.ceil((pct / 100) * len(data)) - 1))
    return data[idx]

groups = defaultdict(list)
for row in rows:
    groups[row["label"]].append(row)

print(f"samples={len(rows)}")
print(f"failures={sum(r['success'].lower() != 'true' for r in rows)}")
print(f"jtl_bytes_on_disk={path.stat().st_size}")

for label, items in sorted(groups.items()):
    elapsed = [int(float(r["elapsed"])) for r in items]
    received = [int(float(r["bytes"])) for r in items]
    sent = [int(float(r["sentBytes"])) for r in items]
    print(
        f"{label}: n={len(items)} "
        f"p50_ms={percentile(elapsed,50)} p95_ms={percentile(elapsed,95)} "
        f"sum_received_bytes={sum(received)} sum_sent_bytes={sum(sent)} "
        f"failures={sum(r['success'].lower() != 'true' for r in items)}"
    )

python tools/analyze_transfer_jtl.py results/lean/results.jtl

Interpret sentBytes/bytes as JMeter sample transfer counters, not raw-file size. The payload manifest and server events provide exact file/body sizes.

16. Analyze server storage/transfer evidence

Save tools/analyze_server_events.py:

import json
import math
import sys
from collections import Counter, defaultdict
from pathlib import Path

path = Path(sys.argv[1])
events = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
if not events:
    raise SystemExit("No server events")

def percentile(values, pct):
    data = sorted(values)
    if not data:
        return 0
    idx = max(0, min(len(data) - 1, math.ceil((pct / 100) * len(data)) - 1))
    return data[idx]

ops = Counter(e["operation"] for e in events)
statuses = Counter(e["status"] for e in events)
print(f"events={len(events)}")
print(f"operations={dict(ops)}")
print(f"statuses={dict(statuses)}")
print(f"peak_storage_bytes={max(e.get('storage_bytes_after',0) for e in events)}")

for operation in ("upload", "download", "delete"):
    items = [e for e in events if e["operation"] == operation]
    wall = [int(e.get("service_wall_ms", 0)) for e in items]
    if items:
        print(
            f"{operation}: n={len(items)} "
            f"p50_service_wall_ms={percentile(wall,50)} "
            f"p95_service_wall_ms={percentile(wall,95)} "
            f"payload_bytes={sum(int(e.get('payload_bytes',0)) for e in items)} "
            f"request_body_bytes={sum(int(e.get('request_content_length',0)) for e in items)}"
        )

python tools/analyze_server_events.py results/server-events.jsonl

Server event evidence separates multipart request-body bytes from file payload bytes, reports upload/download/delete service time, and exposes peak guarded storage.

17. Verify cleanup independently

curl --fail --silent http://127.0.0.1:8018/stats

Require stored_uploads=0 and stored_bytes=0. If not, preserve the orphan evidence before any manual cleanup.

18. Challenge

The application accepts a 10 MiB upload, but the lab generator has limited disk/RAM and the production traffic mix is 90% 100 KiB files, 10% 10 MiB. Should every virtual user upload 10 MiB?

No. Model the observed payload distribution and validate injector/network capacity first. A worst-case-only 10 MiB workload changes bytes/s, storage and generator cost and may answer a stress question rather than the normal workload question.

Knowledge check

Official references and version notes

• JMeter Component Reference — HTTP Request — multipart/form-data, file path/parameter/MIME fields, browser-compatible headers, and Save response as MD5 hash.
• JMeter Component Reference — Save Responses to a file — response-to-disk behavior, filename prefix, numbering, and variable output.
• JMeter Listeners / Result files — CSV fields including bytes/sentBytes, listener memory guidance, response-data retention, and CLI -l behavior.
• JMeter Properties Reference — save-service defaults, httpsampler.max_bytes_to_store_per_request, buffer sizing, and result retention settings.
• JMeter 5.6.3 Changes — expression support for HTTP Request “store as MD5” and current release changes.
• JMeter Best Practices — GUI for authoring/debug and CLI mode for meaningful load.
• Apache JMeter downloads — current stable release and Java requirement.