(t *testing.T, nvs cos.SimpleKVs)
| 100 | } |
| 101 | |
| 102 | func testBundle(t *testing.T, nvs cos.SimpleKVs) { |
| 103 | var ( |
| 104 | numCompleted atomic.Int64 |
| 105 | mmsa, _ = memsys.NewMMSA("bundle") |
| 106 | network = cmn.NetIntraData |
| 107 | trname = "bundle" + nvs["block"] |
| 108 | tss = make([]*httptest.Server, 0, 32) |
| 109 | ) |
| 110 | smap.Tmap = make(cluster.NodeMap, 100) |
| 111 | for i := 0; i < 10; i++ { |
| 112 | ts := httptest.NewServer(objmux) |
| 113 | tss = append(tss, ts) |
| 114 | addTarget(&smap, ts, i) |
| 115 | } |
| 116 | defer func() { |
| 117 | for _, ts := range tss { |
| 118 | ts.Close() |
| 119 | } |
| 120 | mmsa.Terminate(false) |
| 121 | }() |
| 122 | smap.Version = 1 |
| 123 | |
| 124 | receive := func(hdr transport.ObjHdr, objReader io.Reader, err error) error { |
| 125 | if err != nil && !cos.IsEOF(err) { |
| 126 | tassert.CheckFatal(t, err) |
| 127 | } |
| 128 | written, _ := io.Copy(io.Discard, objReader) |
| 129 | cos.Assert(written == hdr.ObjAttrs.Size || hdr.IsUnsized()) |
| 130 | return nil |
| 131 | } |
| 132 | callback := func(_ transport.ObjHdr, _ io.ReadCloser, _ interface{}, _ error) { |
| 133 | numCompleted.Inc() |
| 134 | } |
| 135 | |
| 136 | err := transport.HandleObjStream(trname, receive) // URL = /v1/transport/10G |
| 137 | tassert.CheckFatal(t, err) |
| 138 | defer transport.Unhandle(trname) |
| 139 | |
| 140 | var ( |
| 141 | httpclient = transport.NewIntraDataClient() |
| 142 | sowner = &sowner{} |
| 143 | lsnode = cluster.Snode{DaeID: "local"} |
| 144 | random = newRand(mono.NanoTime()) |
| 145 | wbuf, slab = mmsa.Alloc() |
| 146 | extra = &transport.Extra{Compression: nvs["compression"], MMSA: mmsa} |
| 147 | size, prevsize int64 |
| 148 | multiplier = int(random.Int63()%13) + 4 |
| 149 | num int |
| 150 | usePDU bool |
| 151 | ) |
| 152 | if nvs["compression"] != apc.CompressNever { |
| 153 | v, _ := cos.S2B(nvs["block"]) |
| 154 | cos.Assert(v == cos.MiB*4 || v == cos.MiB || v == cos.KiB*256 || v == cos.KiB*64) |
| 155 | config := cmn.GCO.BeginUpdate() |
| 156 | config.Transport.LZ4BlockMaxSize = cos.Size(v) |
| 157 | cmn.GCO.CommitUpdate(config) |
| 158 | if err := config.Transport.Validate(); err != nil { |
| 159 | tassert.CheckFatal(t, err) |
no test coverage detected