Re: thread resource leaks if thread terminates with uncaught exception

Jens Thiele <[email protected]> writes:

> Jens Thiele <[email protected]> writes:
>
>> 		   (pmap (lambda(i) (http-get-body))
>> 			 (iota 400)
>> 			 :mapper (make-fully-concurrent-mapper
>> 				  0.1 'timeout)))
>
> wanted to improve the pmap method (with fully-concurrent-mapper with
> timeout):
>
> (define-method run-map ((mapper <fully-concurrent-mapper>) proc coll)
>   (let ([unique (list #f)]
>         [ts (map (^e (make-thread (^[] (proc e)))) coll)]
>         [timeout (absolute-time (~ mapper'timeout))]
>         [timeout-val (~ mapper'timeout-val)])
>     (%start-threads ts)
>     (if timeout
>       ($ map (^r (if (and (pair? r) (eq? (car r) unique))
>                    (begin (thread-terminate! (cdr r)) timeout-val)
>                    r))
>          $ map (^t (thread-join! t timeout (cons unique t))) ts)
>       (map thread-join! ts))))
>
> but I am not even sure what should happen if there are multiple uncaught
> exceptions. At the moment the first uncaught exception is raised by
> thread-join!

maybe just keep it like that

but this seems not to be good enough:

(with-module control.pmap
(define-method run-map ((mapper <fully-concurrent-mapper>) proc coll)
  (let* ([unique (list #f)]
         [ts (map (^e (make-thread (^[] (proc e)))) coll)]
	 [cleanup-ts (lambda()
		       (let loop ((ts ts))
			 (unless (null? ts)
			   (set-car! ts #f)
			   (loop (cdr ts)))))]
         [timeout (absolute-time (~ mapper'timeout))]
         [timeout-val (~ mapper'timeout-val)])
    (%start-threads ts)
    (if timeout
      (let1 results (guard (e [(uncaught-exception-condition? e)
			       ;; kill still running threads
			       (for-each thread-terminate! ts)
			       ;; otherwise we get thread died a
			       ;; lonely death with an uncaught
			       ;; exception
			       (for-each
				(lambda(t)
				  (guard
				   (e [(terminated-thread-exception? e)
				       #t]
				      [else
				       ;; #?=e
				       #t
				       ])
				   (thread-join! t)))
				ts)
			       (cleanup-ts)
			       (raise e)])
			   (map (lambda(t)
				  (thread-join! t
						timeout
						(cons unique t)))
				ts))
	(cleanup-ts)
	(map
	 (lambda(r)
	   (if (and (pair? r) (eq? (car r) unique))
	     (begin
	       (thread-terminate! (cdr r))
	       (set-cdr! r #f)
	       timeout-val)
	     r))
	 results))
      (map thread-join! ts))))
)

The complete reverse proxy example that still fails for me:

#!/bin/sh
#| -*- mode: scheme; coding: utf-8; -*-
#export GC_PRINT_STATS=1
exec gosh -I. -- $0 "$@"
|#
(use makiki)
(use text.html-lite)
(use rfc.http)
(use gauche.threads)
(use file.util)
(use gauche.sequence)
(use control.pmap)

(with-module control.pmap
(define-method run-map ((mapper <fully-concurrent-mapper>) proc coll)
  (let* ([unique (list #f)]
         [ts (map (^e (make-thread (^[] (proc e)))) coll)]
	 [cleanup-ts (lambda()
		       (let loop ((ts ts))
			 (unless (null? ts)
			   (set-car! ts #f)
			   (loop (cdr ts)))))]
         [timeout (absolute-time (~ mapper'timeout))]
         [timeout-val (~ mapper'timeout-val)])
    (%start-threads ts)
    (if timeout
      (let1 results (guard (e [(uncaught-exception-condition? e)
			       ;; kill still running threads
			       (for-each thread-terminate! ts)
			       ;; otherwise we get thread died a
			       ;; lonely death with an uncaught
			       ;; exception
			       (for-each
				(lambda(t)
				  (guard
				   (e [(terminated-thread-exception? e)
				       #t]
				      [else
				       ;; #?=e
				       #t
				       ])
				   (thread-join! t)))
				ts)
			       (cleanup-ts)
			       (raise e)])
			   (map (lambda(t)
				  (thread-join! t
						timeout
						(cons unique t)))
				ts))
	(cleanup-ts)
	(map
	 (lambda(r)
	   (if (and (pair? r) (eq? (car r) unique))
	     (begin
	       (thread-terminate! (cdr r))
	       (set-cdr! r #f)
	       timeout-val)
	     r))
	 results))
      (map thread-join! ts))))
)

;; todo: linux specific
(define (num-open-files)
  (guard (e
	  [else
	   +nan.0])
	 (length (directory-list "/proc/self/fd" :children? #t))))

(define (num-threads)
  (guard (e
	  [else
	   +nan.0])
	 (length (directory-list "/proc/self/task" :children? #t))))

(define-http-handler (GET) "/" (^[req app]
				 (receive (status headers body)
				     (http-get "localhost:8081" "/")
				   (respond/ok req body))))

(define (http-get-body)
  (receive (status headers body)
      ;; todo: detect host name and use it to force a real
      ;; dns lookup to occur?
      (http-get "localhost:8081" "/slow")
    body))

(define-http-handler (GET)
  "/timeout"
  (lambda(req app)
    (let1 r (guard (e
		    [(uncaught-exception-condition? e)
		     #?=(uncaught-exception-condition-reason e)]
		    [else #?=e])
		   (pmap (lambda(i) (http-get-body))
			 (iota 400)
			 :mapper (make-fully-concurrent-mapper
				  0.1 'timeout)))
      (cond [(and (list? r) (eq? (car r) 'timeout))
	     (respond/ok req
			 (list
			  (html-doctype)
			  (html:html
			   (html:head (html:title "timeout"))
			   (html:body
			    (html:p "timeout")
			    (html:p
			     (string-append
			      (x->string (num-open-files))
			      " open files"))))))]
	    [(list? r)
	     (car body)]
	    [else
	     (request-error :body r)]))))

(define (main args)
  #?=(sys-getpid)
  (debug-print-width 4000)
  (start-http-server :port 8080 :error-log #t)
  0)

running out of ideas - should I just stop it?

Jens