golang/libnbd_590_aio_copy_test.go

   1 /* libnbd golang tests
   2  * Copyright Red Hat
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Lesser General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Lesser General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Lesser General Public
  15  * License along with this library; if not, write to the Free Software
  16  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  17  */
  18
  19 package libnbd
  20
  21 import "fmt"
  22 import "syscall"
  23 import "testing"
  24
  25 var disk_size = uint(512 * 1024 * 1024)
  26 var bs = uint64(65536)
  27 var max_reads_in_flight = uint(16)
  28 var bytes_read = uint(0)
  29 var bytes_written = uint(0)
  30
  31 /* Functions to handle FdSet.
  32    XXX These probably only work on 64 bit platforms. */
  33 func fdset_set(set *syscall.FdSet, fd int) {
  34         (*set).Bits[fd/64] |= 1 << (uintptr(fd) % 64)
  35 }
  36
  37 func fdset_test(set *syscall.FdSet, fd int) bool {
  38         return (*set).Bits[fd/64]&(1<<(uintptr(fd)%64)) != 0
  39 }
  40
  41 /* Functions to test socket direction. */
  42 func dir_is_read(h *Libnbd) bool {
  43         dir, _ := h.AioGetDirection()
  44         return (uint32(dir) & AIO_DIRECTION_READ) != 0
  45 }
  46 func dir_is_write(h *Libnbd) bool {
  47         dir, _ := h.AioGetDirection()
  48         return (uint32(dir) & AIO_DIRECTION_WRITE) != 0
  49 }
  50
  51 /* Queue of writes. */
  52 type wbuf struct {
  53         buf    AioBuffer
  54         offset uint64
  55 }
  56
  57 var writes []wbuf
  58
  59 /* Called whenever any asynchronous pread command from
  60    the source has completed. */
  61 func read_completed(buf AioBuffer, offset uint64) int {
  62         bytes_read += buf.Size
  63         /* Move the AIO buffer to the write queue. */
  64         writes = append(writes, wbuf{buf, offset})
  65         /* Returning 1 means the command is automatically retired. */
  66         return 1
  67 }
  68
  69 /* Called whenever any asynchronous pwrite command to the
  70    destination has completed. */
  71 func write_completed(buf AioBuffer) int {
  72         bytes_written += buf.Size
  73         /* Now we have to manually free the AIO buffer. */
  74         buf.Free()
  75         /* Returning 1 means the command is automatically retired. */
  76         return 1
  77 }
  78
  79 /* Copy between two libnbd handles using aynchronous I/O (AIO). */
  80 func asynch_copy(t *testing.T, src *Libnbd, dst *Libnbd) {
  81         size, _ := dst.GetSize()
  82
  83         /* This is our reading position in the source. */
  84         soff := uint64(0)
  85
  86         for {
  87                 /* Number of commands in flight on source and dest handles. */
  88                 src_in_flight, _ := src.AioInFlight()
  89                 dst_in_flight, _ := dst.AioInFlight()
  90
  91                 /* We're finished when we've read everything from the
  92                    source and there are no commands in flight. */
  93                 if soff >= size && src_in_flight == 0 &&
  94                         dst_in_flight == 0 {
  95                         break
  96                 }
  97
  98                 /* If we're able to submit more reads from the
  99                    source then do it now. */
 100                 if soff < size && src_in_flight < max_reads_in_flight {
 101                         n := bs
 102                         if n > size-soff {
 103                                 n = size - soff
 104                         }
 105                         buf := MakeAioBuffer(uint(n))
 106                         soff_copy := soff
 107                         var optargs AioPreadOptargs
 108                         optargs.CompletionCallbackSet = true
 109                         optargs.CompletionCallback = func(error *int) int {
 110                                 if *error != 0 {
 111                                         err := syscall.Errno(*error).Error()
 112                                         panic(err)
 113                                 }
 114                                 return read_completed(buf, soff_copy)
 115                         }
 116                         src.AioPread(buf, soff, &optargs)
 117                         soff += n
 118                 }
 119
 120                 /* If there are any write commands waiting to
 121                    be issued, send them now. */
 122                 for _, wb := range writes {
 123                         var optargs AioPwriteOptargs
 124                         optargs.CompletionCallbackSet = true
 125                         optargs.CompletionCallback = func(error *int) int {
 126                                 if *error != 0 {
 127                                         err := syscall.Errno(*error).Error()
 128                                         panic(err)
 129                                 }
 130                                 return write_completed(wb.buf)
 131                         }
 132                         dst.AioPwrite(wb.buf, wb.offset, &optargs)
 133                 }
 134                 writes = writes[:0]
 135
 136                 /* Now poll the file descriptors. */
 137                 nfd := 1
 138                 sfd, err := src.AioGetFd()
 139                 if err != nil {
 140                         t.Fatalf("src.AioGetFd: %s", err)
 141                 }
 142                 if sfd >= nfd {
 143                         nfd = sfd + 1
 144                 }
 145                 dfd, err := dst.AioGetFd()
 146                 if err != nil {
 147                         t.Fatalf("dst.AioGetFd: %s", err)
 148                 }
 149                 if dfd >= nfd {
 150                         nfd = dfd + 1
 151                 }
 152                 var rfds syscall.FdSet
 153                 if dir_is_read(src) {
 154                         fdset_set(&rfds, sfd)
 155                 }
 156                 if dir_is_read(dst) {
 157                         fdset_set(&rfds, dfd)
 158                 }
 159                 var wfds syscall.FdSet
 160                 if dir_is_write(src) {
 161                         fdset_set(&wfds, sfd)
 162                 }
 163                 if dir_is_write(dst) {
 164                         fdset_set(&wfds, dfd)
 165                 }
 166                 for {
 167                         _, err = syscall.Select(nfd, &rfds, &wfds, nil, nil)
 168                         if err != syscall.EINTR {
 169                                 break
 170                         }
 171                 }
 172                 if err != nil {
 173                         t.Fatalf("select: %s", err)
 174                 }
 175
 176                 if fdset_test(&rfds, sfd) && dir_is_read(src) {
 177                         src.AioNotifyRead()
 178                 } else if fdset_test(&wfds, sfd) && dir_is_write(src) {
 179                         src.AioNotifyWrite()
 180                 } else if fdset_test(&rfds, dfd) && dir_is_read(dst) {
 181                         dst.AioNotifyRead()
 182                 } else if fdset_test(&wfds, dfd) && dir_is_write(dst) {
 183                         dst.AioNotifyWrite()
 184                 }
 185         }
 186 }
 187
 188 func Test590AioCopy(t *testing.T) {
 189         src, err := Create()
 190         if err != nil {
 191                 t.Fatalf("could not create handle: %s", err)
 192         }
 193         defer src.Close()
 194         src.SetHandleName("src")
 195
 196         var dst *Libnbd
 197         dst, err = Create()
 198         if err != nil {
 199                 t.Fatalf("could not create handle: %s", err)
 200         }
 201         defer dst.Close()
 202         dst.SetHandleName("dst")
 203
 204         err = src.ConnectCommand([]string{
 205                 "nbdkit", "-s", "--exit-with-parent", "-r",
 206                 "pattern", fmt.Sprintf("size=%d", disk_size),
 207         })
 208         if err != nil {
 209                 t.Fatalf("could not connect: %s", err)
 210         }
 211
 212         err = dst.ConnectCommand([]string{
 213                 "nbdkit", "-s", "--exit-with-parent",
 214                 "memory", fmt.Sprintf("size=%d", disk_size),
 215         })
 216         if err != nil {
 217                 t.Fatalf("could not connect: %s", err)
 218         }
 219
 220         asynch_copy(t, src, dst)
 221         if bytes_read != disk_size {
 222                 t.Fatalf("bytes_read != disk_size")
 223         }
 224         if bytes_written != disk_size {
 225                 t.Fatalf("bytes_written != disk_size")
 226         }
 227 }