Merge pull request #28 from kylelemons/async_transfers_everywhere

Use async transfer API everywhere

usb/config.go

@@ -38,14 +38,17 @@ func (e EndpointInfo) Number() int {
 	return int(e.Address) & ENDPOINT_NUM_MASK
 }
 
+func (e EndpointInfo) TransferType() TransferType {
+	return TransferType(e.Attributes) & TRANSFER_TYPE_MASK
+}
+
 func (e EndpointInfo) Direction() EndpointDirection {
 	return EndpointDirection(e.Address) & ENDPOINT_DIR_MASK
 }
 
 func (e EndpointInfo) String() string {
 	return fmt.Sprintf("Endpoint %d %-3s %s - %s %s [%d %d]",
-		e.Number(), e.Direction(),
+		e.Number(), e.Direction(), e.TransferType(),
-		TransferType(e.Attributes)&TRANSFER_TYPE_MASK,
 		IsoSyncType(e.Attributes)&ISO_SYNC_TYPE_MASK,
 		IsoUsageType(e.Attributes)&ISO_USAGE_TYPE_MASK,
 		e.MaxPacketSize, e.MaxIsoPacket,
@@ -135,7 +138,7 @@ func newConfig(dev *C.libusb_device, cfg *C.struct_libusb_config_descriptor) Con
 					RefreshRate:   uint8(end.bRefresh),
 					SynchAddress:  uint8(end.bSynchAddress),
 				}
-				if TransferType(ei.Attributes)&TRANSFER_TYPE_MASK == TRANSFER_TYPE_ISOCHRONOUS {
+				if ei.TransferType() == TRANSFER_TYPE_ISOCHRONOUS {
 					// bits 0-10 identify the packet size, bits 11-12 are the number of additional transactions per microframe.
 					// Don't use libusb_get_max_iso_packet_size, as it has a bug where it returns the same value
 					// regardless of alternative setting used, where different alternative settings might define different

usb/device.go

@@ -151,16 +151,6 @@ func (d *Device) OpenEndpoint(conf, iface, setup, epoint uint8) (Endpoint, error
 					}
 					end.InterfaceSetup = s
 					end.EndpointInfo = e
-					switch tt := TransferType(e.Attributes) & TRANSFER_TYPE_MASK; tt {
-					case TRANSFER_TYPE_BULK:
-						end.xfer = bulk_xfer
-					case TRANSFER_TYPE_INTERRUPT:
-						end.xfer = interrupt_xfer
-					case TRANSFER_TYPE_ISOCHRONOUS:
-						end.xfer = isochronous_xfer
-					default:
-						return nil, fmt.Errorf("usb: %s transfer is unsupported", tt)
-					}
 					goto found
 				}
 				return nil, fmt.Errorf("usb: unknown endpoint %02x", epoint)

usb/endpoint.go

@@ -20,9 +20,8 @@ import "C"
 
 import (
 	"fmt"
-	"reflect"
+	"log"
 	"time"
-	"unsafe"
 )
 
 type Endpoint interface {
@@ -36,7 +35,6 @@ type endpoint struct {
 	*Device
 	InterfaceSetup
 	EndpointInfo
-	xfer func(*endpoint, []byte, time.Duration) (int, error)
 }
 
 func (e *endpoint) Read(buf []byte) (int, error) {
@@ -44,7 +42,7 @@ func (e *endpoint) Read(buf []byte) (int, error) {
 		return 0, fmt.Errorf("usb: read: not an IN endpoint")
 	}
 
-	return e.xfer(e, buf, e.ReadTimeout)
+	return e.transfer(buf, e.ReadTimeout)
 }
 
 func (e *endpoint) Write(buf []byte) (int, error) {
@@ -52,50 +50,33 @@ func (e *endpoint) Write(buf []byte) (int, error) {
 		return 0, fmt.Errorf("usb: write: not an OUT endpoint")
 	}
 
-	return e.xfer(e, buf, e.WriteTimeout)
+	return e.transfer(buf, e.WriteTimeout)
 }
 
 func (e *endpoint) Interface() InterfaceSetup { return e.InterfaceSetup }
 func (e *endpoint) Info() EndpointInfo        { return e.EndpointInfo }
 
-// TODO(kevlar): (*Endpoint).Close
+func (e *endpoint) transfer(buf []byte, timeout time.Duration) (int, error) {
-
-func bulk_xfer(e *endpoint, buf []byte, timeout time.Duration) (int, error) {
 	if len(buf) == 0 {
 		return 0, nil
 	}
 
-	data := (*reflect.SliceHeader)(unsafe.Pointer(&buf)).Data
+	tt := e.TransferType()
-
+	t, err := newUSBTransfer(e.Device.handle, e.EndpointInfo, buf, timeout)
-	var cnt C.int
+	if err != nil {
-	if errno := C.libusb_bulk_transfer(
+		return 0, err
-		e.handle,
-		C.uchar(e.Address),
-		(*C.uchar)(unsafe.Pointer(data)),
-		C.int(len(buf)),
-		&cnt,
-		C.uint(timeout/time.Millisecond)); errno < 0 {
-		return 0, usbError(errno)
 	}
-	return int(cnt), nil
+	defer t.free()
-}
+
-
+	if err := t.submit(); err != nil {
-func interrupt_xfer(e *endpoint, buf []byte, timeout time.Duration) (int, error) {
+		log.Printf("bulk: %s failed to submit: %s", tt, err)
-	if len(buf) == 0 {
+		return 0, err
-		return 0, nil
+	}
-	}
+
-
+	n, err := t.wait()
-	data := (*reflect.SliceHeader)(unsafe.Pointer(&buf)).Data
+	if err != nil {
-
+		log.Printf("bulk: %s failed: %s", tt, err)
-	var cnt C.int
+		return 0, err
-	if errno := C.libusb_interrupt_transfer(
+	}
-		e.handle,
+	return n, err
-		C.uchar(e.Address),
-		(*C.uchar)(unsafe.Pointer(data)),
-		C.int(len(buf)),
-		&cnt,
-		C.uint(timeout/time.Millisecond)); errno < 0 {
-		return 0, usbError(errno)
-	}
-	return int(cnt), nil
 }

usb/iso.go

@@ -1,150 +0,0 @@
-// Copyright 2013 Google Inc.  All rights reserved.
-// Copyright 2016 the gousb Authors.  All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package usb
-
-/*
-#include <libusb.h>
-
-int submit(struct libusb_transfer *xfer);
-void print_xfer(struct libusb_transfer *xfer);
-int extract_data(struct libusb_transfer *xfer, void *data, int max, unsigned char *status);
-*/
-import "C"
-
-import (
-	"fmt"
-	"log"
-	"time"
-	"unsafe"
-)
-
-//export iso_callback
-func iso_callback(cptr unsafe.Pointer) {
-	ch := *(*chan struct{})(cptr)
-	close(ch)
-}
-
-func (end *endpoint) allocTransfer(maxLen int) *Transfer {
-	isoPacketSize := end.EndpointInfo.MaxIsoPacket
-	// the larget the input buffer, the more packets we use in a single
-	// transfer.
-	numIsoPackets := maxLen / int(isoPacketSize)
-	if numIsoPackets*int(isoPacketSize) < maxLen {
-		numIsoPackets++
-	}
-	xfer := C.libusb_alloc_transfer(C.int(numIsoPackets))
-	if xfer == nil {
-		log.Printf("usb: transfer allocation failed?!")
-		return nil
-	}
-
-	buf := make([]byte, numIsoPackets*int(end.EndpointInfo.MaxIsoPacket))
-	done := make(chan struct{}, 1)
-
-	xfer.dev_handle = end.Device.handle
-	xfer.endpoint = C.uchar(end.Address)
-	xfer._type = C.LIBUSB_TRANSFER_TYPE_ISOCHRONOUS
-
-	xfer.buffer = (*C.uchar)((unsafe.Pointer)(&buf[0]))
-	xfer.length = C.int(len(buf))
-	xfer.num_iso_packets = C.int(numIsoPackets)
-
-	C.libusb_set_iso_packet_lengths(xfer, C.uint(end.EndpointInfo.MaxIsoPacket))
-	/*
-		pkts := *(*[]C.struct_libusb_packet_descriptor)(unsafe.Pointer(&reflect.SliceHeader{
-			Data: uintptr(unsafe.Pointer(&xfer.iso_packet_desc)),
-			Len:  iso_packets,
-			Cap:  iso_packets,
-		}))
-	*/
-
-	t := &Transfer{
-		xfer: xfer,
-		done: done,
-		buf:  buf,
-	}
-	xfer.user_data = (unsafe.Pointer)(&t.done)
-
-	return t
-}
-
-type Transfer struct {
-	xfer *C.struct_libusb_transfer
-	pkts []*C.struct_libusb_packet_descriptor
-	done chan struct{}
-	buf  []byte
-}
-
-func (t *Transfer) Submit(timeout time.Duration) error {
-	//log.Printf("iso: submitting %#v", t.xfer)
-	t.xfer.timeout = C.uint(timeout / time.Millisecond)
-	if errno := C.submit(t.xfer); errno < 0 {
-		return usbError(errno)
-	}
-	return nil
-}
-
-func (t *Transfer) Wait(b []byte) (n int, err error) {
-	select {
-	case <-time.After(10 * time.Second):
-		return 0, fmt.Errorf("wait timed out after 10s")
-	case <-t.done:
-	}
-	// Non-iso transfers:
-	//n = int(t.xfer.actual_length)
-	//copy(b, ((*[1 << 16]byte)(unsafe.Pointer(t.xfer.buffer)))[:n])
-
-	//C.print_xfer(t.xfer)
-	/*
-		buf, offset := ((*[1 << 16]byte)(unsafe.Pointer(t.xfer.buffer))), 0
-		for i, pkt := range *t.pkts {
-			log.Printf("Type is %T", t.pkts)
-			n += copy(b[n:], buf[offset:][:pkt.actual_length])
-			offset += pkt.Length
-			if pkt.status != 0 && err == nil {
-				err = error(TransferStatus(pkt.status))
-			}
-		}
-	*/
-	var status uint8
-	n = int(C.extract_data(t.xfer, unsafe.Pointer(&b[0]), C.int(len(b)), (*C.uchar)(unsafe.Pointer(&status))))
-	if status != 0 {
-		err = TransferStatus(status)
-	}
-	return n, err
-}
-
-func (t *Transfer) Close() error {
-	C.libusb_free_transfer(t.xfer)
-	return nil
-}
-
-func isochronous_xfer(e *endpoint, buf []byte, timeout time.Duration) (int, error) {
-	t := e.allocTransfer(len(buf))
-	defer t.Close()
-
-	if err := t.Submit(timeout); err != nil {
-		log.Printf("iso: xfer failed to submit: %s", err)
-		return 0, err
-	}
-
-	n, err := t.Wait(buf)
-	if err != nil {
-		log.Printf("iso: xfer failed: %s", err)
-		return 0, err
-	}
-	return n, err
-}

usb/transfer.c

@@ -18,25 +18,15 @@
 #include <string.h>
 
 void print_xfer(struct libusb_transfer *xfer);
-void iso_callback(void *);
+void xfer_callback(void *);
 
 void callback(struct libusb_transfer *xfer) {
-	//printf("Callback!\n");
+	xfer_callback(xfer->user_data);
-	//print_xfer(xfer);
-	iso_callback(xfer->user_data);
 }
 
 int submit(struct libusb_transfer *xfer) {
 	xfer->callback = (libusb_transfer_cb_fn)(&callback);
 	xfer->status = -1;
-	//print_xfer(xfer);
-	//printf("Transfer submitted\n");
-
-	/* fake
-	strcpy(xfer->buffer, "hello");
-	xfer->actual_length = 5;
-	callback(xfer);
-	return 0; */
 	return libusb_submit_transfer(xfer);
 }
 
@@ -64,35 +54,27 @@ void print_xfer(struct libusb_transfer *xfer) {
 	}
 }
 
-int extract_data(struct libusb_transfer *xfer, void *raw, int max, unsigned char *status) {
+// compact the data in an isochronous transfer. The contents of individual
+// iso packets are shifted left, so that no gaps are left between them.
+// Status is set to the first non-zero status of an iso packet.
+int compact_iso_data(struct libusb_transfer *xfer, unsigned char *status) {
 	int i;
-	int copied = 0;
+	int sum = 0;
 	unsigned char *in = xfer->buffer;
-	unsigned char *out = raw;
+	unsigned char *out = xfer->buffer;
 	for (i = 0; i < xfer->num_iso_packets; i++) {
 		struct libusb_iso_packet_descriptor pkt = xfer->iso_packet_desc[i];
-
+		if (pkt.status != 0) {
-		// Copy the data
+		    *status = pkt.status;
-		int len = pkt.actual_length;
-		if (copied + len > max) {
-			len = max - copied;
-		}
-		memcpy(out, in, len);
-		copied += len;
-
-		// Increment offsets
-		in += pkt.length;
-		out += len;
-
-        if (copied == max) {
 			break;
 		}
-
+		// Copy the data
-		// Extract first error
+		int len = pkt.actual_length;
-		if (pkt.status == 0 || *status != 0) {
+		memmove(out, in, len);
-			continue;
+		// Increment offsets
+		sum += len;
+		in += pkt.length;
+		out += len;
 	}
-		*status = pkt.status;
+	return sum;
-	}
-	return copied;
 }

usb/transfer.go

@@ -0,0 +1,129 @@
+// Copyright 2016 the gousb Authors.  All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package usb
+
+/*
+#include <libusb.h>
+
+int compact_iso_data(struct libusb_transfer *xfer, unsigned char *status);
+int submit(struct libusb_transfer *xfer);
+*/
+import "C"
+
+import (
+	"fmt"
+	"time"
+	"unsafe"
+)
+
+//export xfer_callback
+func xfer_callback(cptr unsafe.Pointer) {
+	ch := *(*chan struct{})(cptr)
+	close(ch)
+}
+
+type usbTransfer struct {
+	// xfer is the allocated libusb_transfer.
+	xfer *C.struct_libusb_transfer
+	// buf is the buffer allocated for the transfer. Both buf and xfer.buffer
+	// point to the same piece of memory.
+	buf []byte
+	// done is blocking until the transfer is complete and data and transfer
+	// status are available.
+	done chan struct{}
+}
+
+// submits the transfer. After submit() the transfer is in flight and is owned by libusb.
+// It's not safe to access the contents of the transfer until wait() returns.
+// Once wait() returns, it's ok to re-use the same transfer structure by calling submit() again.
+func (t *usbTransfer) submit() error {
+	t.done = make(chan struct{})
+	t.xfer.user_data = (unsafe.Pointer)(&t.done)
+	if errno := C.submit(t.xfer); errno < 0 {
+		return usbError(errno)
+	}
+	return nil
+}
+
+// wait waits for libusb to signal the release of transfer data.
+// After wait returns, the transfer contents are safe to access
+// via t.buf. The number returned by wait indicates how many bytes
+// of the buffer were read or written by libusb, and it can be
+// smaller than the length of t.buf.
+func (t *usbTransfer) wait() (n int, err error) {
+	select {
+	case <-time.After(10 * time.Second):
+		return 0, fmt.Errorf("wait timed out after 10s")
+	case <-t.done:
+	}
+	var status TransferStatus
+	switch TransferType(t.xfer._type) {
+	case TRANSFER_TYPE_ISOCHRONOUS:
+		n = int(C.compact_iso_data(t.xfer, (*C.uchar)(unsafe.Pointer(&status))))
+	default:
+		n = int(t.xfer.length)
+		status = TransferStatus(t.xfer.status)
+	}
+	if status != LIBUSB_TRANSFER_COMPLETED {
+		return 0, status
+	}
+	return n, err
+}
+
+// free releases the memory allocated for the transfer.
+// free should be called only if the transfer is not used by libusb,
+// i.e. it should not be called after submit() and before wait() returns.
+func (t *usbTransfer) free() error {
+	C.libusb_free_transfer(t.xfer)
+	t.xfer = nil
+	t.buf = nil
+	t.done = nil
+	return nil
+}
+
+type deviceHandle *C.libusb_device_handle
+
+// newUSBTransfer allocates a new transfer structure for communication with a
+// given device/endpoint, with buf as the underlying transfer buffer.
+func newUSBTransfer(dev deviceHandle, ei EndpointInfo, buf []byte, timeout time.Duration) (*usbTransfer, error) {
+	var isoPackets int
+	tt := ei.TransferType()
+	if tt == TRANSFER_TYPE_ISOCHRONOUS {
+		isoPackets = len(buf) / int(ei.MaxIsoPacket)
+	}
+
+	xfer := C.libusb_alloc_transfer(C.int(isoPackets))
+	if xfer == nil {
+		return nil, fmt.Errorf("libusb_alloc_transfer(%d) failed", isoPackets)
+	}
+
+	xfer.dev_handle = dev
+	xfer.timeout = C.uint(timeout / time.Millisecond)
+	xfer.endpoint = C.uchar(ei.Address)
+	xfer._type = C.uchar(tt)
+
+	xfer.buffer = (*C.uchar)((unsafe.Pointer)(&buf[0]))
+	xfer.length = C.int(len(buf))
+
+	if tt == TRANSFER_TYPE_ISOCHRONOUS {
+		xfer.num_iso_packets = C.int(isoPackets)
+		C.libusb_set_iso_packet_lengths(xfer, C.uint(ei.MaxIsoPacket))
+	}
+
+	return &usbTransfer{
+		xfer: xfer,
+		buf:  buf,
+	}, nil
+}