Le 2015-12-06 21:38, Marko Rauhamaa a écrit :
Guile 2.0.11 provides:
get-bytevector-n
get-bytevector-n!
get-bytevector-some
Of these, only get-bytevector-some seems to behave as expected when the
port is nonblocking:
First thanks for pointing the correct way to setup a non-blocking port.
I was missing that bit in my code.
I can't make work `get-bytevector-some` with my code. It only returns a
single cell bv.
Instead I use `char-ready?`. Here is the definition of `recv-some`:
```
(define (recv-some port)
(let next ((out '()))
(if (char-ready? port)
(let ((byte (get-u8 port)))
(if (eof-object? byte)
(u8-list->bytevector (reverse out))
(next (cons byte out))))
(u8-list->bytevector (reverse out)))))
```
FWIW, i also have a send-all procedure that works around the fact that
the socket is non blocking:
```
(define (send-all socket message cc)
(let loop ((message message))
(let* ((count (send socket message))
(message (bytevector-drop message count)))
(if (eq? (bytevector-length message) 0)
(cc)
(loop-add-writer socket (lambda () (loop message)))))))
```
`loop-add-writer` register a thunk to be run when the socket is ready
for writes.
I attached tests files.
;; Copyright (C) 2015 Amirouche Boubekki <[email protected]>
;; This library is free software; you can redistribute it and/or
;; modify it under the terms of the GNU Lesser General Public
;; License as published by the Free Software Foundation; either
;; version 3 of the License, or (at your option) any later version.
;;
;; This library is distributed in the hope that it will be useful,
;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
;; Lesser General Public License for more details.
;;
;; You should have received a copy of the GNU Lesser General Public
;; License along with this library; if not, write to the Free Software
;; Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
;; 02110-1301 USA
(define-module (async))
(use-modules (srfi srfi-69)) ;; hash-table
(use-modules (ice-9 q))
(use-modules (ice-9 match))
(use-modules (ice-9 rdelim))
(use-modules (srfi srfi-9))
;; srfi-queue :o)
(define make-queue make-q)
(define push! enq!)
(define pop! deq!)
(define empty? q-empty?)
(define-record-type <loop>
(make-loop running tasks readers writers)
loop?
(running loop-running loop-running!)
(tasks loop-tasks loop-tasks!)
(readers loop-readers loop-readers!)
(writers loop-writers loop-writers!))
(define loop (make-loop #false (make-queue) (make-hash-table)
(make-hash-table)))
(define-public *loop* (make-fluid loop))
(define-public (loop-call-later thunk) ;; support delay
(let ((loop (fluid-ref *loop*)))
(push! (loop-tasks loop) thunk)))
(define (loop-add-reader port callback)
(let* ((loop (fluid-ref *loop*))
(readers (loop-readers loop))
(queue (hash-table-ref readers port make-queue)))
(push! queue callback)
(hash-table-set! readers port queue)))
(define (loop-add-writer port callback)
(let* ((loop (fluid-ref *loop*))
(writers (loop-writers loop))
(queue (hash-table-ref writers port make-queue)))
(push! queue callback)
(hash-table-set! writers port queue)))
(define (call-read-callback port)
(let* ((loop (fluid-ref *loop*))
(queue (hash-table-ref (loop-readers loop) port)))
(when (and queue (not (empty? queue)))
(let ((callback (pop! queue)))
(when (empty? queue)
(hash-table-delete! (loop-readers loop) port))
(callback)))))
(define (call-write-callback port)
(let* ((loop (fluid-ref *loop*))
(queue (hash-table-ref (loop-writers loop) port)))
(when (and queue (not (empty? queue)))
(let ((callback (pop! queue)))
(when (empty? queue)
(hash-table-delete! (loop-writers loop) port))
(callback)))))
(define (loop-run-once)
(let ((loop (fluid-ref *loop*))
(readers (hash-table-keys (loop-readers loop)))
(writers (hash-table-keys (loop-writers loop))))
;; first poll ready ports
(match (select readers writers '() 0) ;; FIXME: replace 0 with time for
next task
((to-read to-write _)
(for-each call-read-callback to-read)
(for-each call-write-callback to-write)))
;; execute tasks
(while (not (empty? (loop-tasks loop)))
((pop! (loop-tasks loop))))))
(define-public (loop-run-forever)
(let* ((loop (fluid-ref *loop*)))
(loop-running! loop #true)
(while (loop-running loop)
(call-with-prompt 'loop
loop-run-once
(lambda (cc callback) (callback cc))))))
(define-public (loop-running?)
(let ((loop (fluid-ref *loop*)))
(loop-running loop)))
(define-public (loop-stop!)
(let ((loop (fluid-ref *loop*)))
(loop-running! loop #false)))
;; non-blocking replacement for read/write procedures
(use-modules (ice-9 binary-ports))
(define (recv-some port)
(let next ((out '()))
(if (char-ready? port)
(let ((byte (get-u8 port)))
(if (eof-object? byte)
(pk 'eof (u8-list->bytevector (reverse out)))
(next (cons byte out))))
(pk '!ready (u8-list->bytevector (reverse out))))))
(define-public (recv-some/ socket)
(abort-to-prompt 'loop (lambda (cc) (loop-add-reader socket (lambda () (cc
(recv-some socket)))))))
(define-public (get-bytevector-some/ socket)
(abort-to-prompt 'loop (lambda (cc) (loop-add-reader socket (lambda () (cc
(get-bytevector-some socket)))))))
(define-public (accept/ socket)
(abort-to-prompt 'loop (lambda (cc) (loop-add-reader socket (lambda () (cc
(accept socket)))))))
(use-modules (rnrs bytevectors))
(define (bytevector-drop bv count)
(u8-list->bytevector (list-tail (bytevector->u8-list bv) count)))
(define (send-all socket message cc)
(let loop ((message message))
(let* ((count (send socket message))
(message (bytevector-drop message count)))
(if (eq? (bytevector-length message) 0)
(cc)
(loop-add-writer socket (lambda () (loop message)))))))
(define-public (send-all/ socket message)
(abort-to-prompt 'loop (lambda (cc) (loop-add-writer socket (lambda ()
(send-all socket message cc))))))
;; XXX: those are blocking anyway
;; (define-public (read/ socket)
;; (abort-to-prompt 'loop (lambda (cc) (loop-add-reader socket (lambda () (cc
(read socket)))))))
;; (define-public (write/ message socket)
;; (abort-to-prompt 'loop (lambda (cc) (loop-add-writer socket (lambda () (cc
(write message socket)))))))
;; (define-public (display/ message socket)
;; (abort-to-prompt 'loop (lambda (cc) (loop-add-writer socket (lambda () (cc
(display message socket)))))))
;; (define-public (read-line/ socket)
;; (abort-to-prompt 'loop (lambda (cc) (loop-add-writer socket (lambda () (cc
(read-line socket)))))))
;;; basic TCP sockets
(define make-socket socket)
(define-public (make-client-socket port)
(let ((socket (make-socket PF_INET SOCK_STREAM 0)))
(connect socket AF_INET INADDR_LOOPBACK port)
(fcntl socket F_SETFL (logior O_NONBLOCK (fcntl socket F_GETFL)))
socket))
(define-public (make-server-socket port)
(let ((socket (make-socket PF_INET SOCK_STREAM 0)))
(bind socket (make-socket-address AF_INET INADDR_ANY port))
(listen socket 128)
(fcntl socket F_SETFL (logior O_NONBLOCK (fcntl socket F_GETFL)))
socket))
(define-module (client))
(use-modules (rnrs bytevectors))
(use-modules (async))
(setlocale LC_ALL "")
(define (scm->string scm)
(call-with-output-string
(lambda (port)
(write scm port))))
(define scm->bv (compose string->utf8 scm->string))
(define (client)
(define socket (make-client-socket 12345))
(send-all/ socket (string->utf8 "héllo world"))
(send-all/ socket #vu8(0))
(send-all/ socket (string->utf8 "you are free"))
(send-all/ socket #vu8(0))
(close socket)
(loop-stop!))
(loop-call-later client)
(loop-run-forever)
(define-module (server))
(use-modules (rnrs bytevectors))
(use-modules (async))
(setlocale LC_ALL "")
;; very simple server socket helper
(define (server)
(let* ((sock (make-server-socket 12345))
(client (car (accept/ sock))))
(pk (utf8->string (recv-some/ client)))
(pk (utf8->string (recv-some/ client)))
(close client)
(loop-stop!)))
(loop-call-later server)
(loop-run-forever)
