From c7e4bc0fd0ddf89ba0177f537cd45084724bf209 Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 4 Nov 2021 12:44:00 -0500 Subject: [PATCH] refactoring experiment I'm not actually sold on this, but I'm not entirely unsold on it. It seems like it does reduce the amount of duplication a lot, but also it's sort of a mess. In the process, noticed that it makes more sense to grab the whole cluster rather than just the nodes for an arbitrary shard for the shard==^0 case, because then if we have an API (but no Qcx), we can be reasonably confident that we'll be able to pick the local node for loopback even if we aren't using the API directly. Have thought about whether we should create our own Qcx in cases like that but I really don't like the idea of automatically creating a Qcx. --- http/client.go | 275 ++++++++++++++++++++----------------------------- 1 file changed, 114 insertions(+), 161 deletions(-) diff --git a/http/client.go b/http/client.go index 12f87d160..628c2a390 100644 --- a/http/client.go +++ b/http/client.go @@ -633,6 +633,109 @@ func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, in return nil } +// importHelper is an experiment to see whether SonarCloud's code duplication +// complaints make sense to address in this context, given the impracticality +// of refactoring ImportRequest/ImportValueRequest right now. The process +// function exists because we would use either api.ImportValueWithTx or +// api.ImportWithTx, passing it the actual underlying-type of req, but doing +// that in here with a type switch seems messy. Similarly, index/field/shard +// exist because we can't access those members of the two slightly different +// structs. +func (c *InternalClient) importHelper(ctx context.Context, req pilosa.Message, process func() error, index string, field string, shard uint64, options *pilosa.ImportOptions) error { + // If we don't actually know what shards we're sending to, and we have + // a local API and a qcx, we'll have a process function that uses the local + // API. Otherwise, even if we have an API + var nodes []*topology.Node + var err error + if shard != ^uint64(0) { + // we need a list of nodes specific to this shard. + nodes, err = c.FragmentNodes(ctx, index, shard) + if err != nil { + return errors.Errorf("shard nodes: %s", err) + } + } else { + // We don't know what shard to use, any shard is fine, local host + // is better if available. + if process != nil { + // skip the HTTP round-trip if we can. + err = process() + // Note that Wrap(nil, ...) is still nil. + return errors.Wrap(err, "local import") + } + // get the complete list of nodes, so if we have an API, we can + // pick our local node and probably avoid actually sending the http + // request over the wire, even though we still have to go through + // the http interface. + nodes, err = c.Nodes(ctx) + } + + // "us" is a usable local node if any, "them" is every node that we need + // to process which isn't that node. We start out with us == nil and + // them = the whole set of nodes. + var us *topology.Node + var them []*topology.Node = nodes + + // If we have an API, we know what node we are. Even if we don't have + // a Qcx, we still care, because looping back to the local node will + // be faster than going to another node. + if c.api != nil { + myID := c.api.NodeID() + for i, node := range nodes { + if myID == node.ID { + // swap our node into the first position + nodes[i], nodes[0] = nodes[0], nodes[i] + // If we have a qcx, we'll treat our node even MORE + // specially. + if process != nil { + us, them = nodes[0], nodes[1:] + } + break + } + } + } + + // If we had a valid API and Qcx, and shard was ^0, we'd have handled + // it previously. So if we get here, we don't have both a Qcx and an API, + // but we might have an API, in which case we'll have shuffled our node + // into the first position. Otherwise we're just taking whatever the first + // node is. + if shard == ^uint64(0) { + them = them[:1] + } + + // We handle remote nodes first, for two distinct reasons. One is that + // the local API ImportWithTx is allowed to modify its inputs, and if we + // ran that before serializing, we'd get corrupt data serialized. + // The other is that if we were to hold a write lock that started with + // that import and ended when we hit the end of our Qcx, we wouldn't want + // to hold it during all our requests to the remote nodes. + if len(them) > 0 { + buf, err := c.serializer.Marshal(req) + if err != nil { + return errors.Errorf("marshal import request: %s", err) + } + // We process remote nodes first so we won't be actually holding our + // write lock yet, in theory. This doesn't actually matter yet, but is + // helpful for future planned refactoring. + for _, node := range them { + if err = c.importNode(ctx, node, index, field, buf, options); err != nil { + return errors.Wrap(err, "remote import") + } + } + } + + // Write to the local node if we have one. + if us != nil { + // WARNING: ImportWithTx can alter its inputs. However, we can + // only ever do this once, and if we're going to need a marshalled + // form, we already made it. + if err = process(); err != nil { + return errors.Wrap(err, "local import after remote imports") + } + } + return nil +} + // Import imports values using an ImportRequest, whether or not it's keyed. // It may modify the contents of req. // @@ -652,88 +755,13 @@ func (c *InternalClient) Import(ctx context.Context, qcx *pilosa.Qcx, req *pilos if req.ColumnKeys != nil { req.Shard = ^uint64(0) } - // If we don't actually know what shards we're sending to, and we have - // a local API, we can short-circuit: We just send this request to the - // local API, which will break it out and call its defaultClient for - // the data for individual shards. - var nodes []*topology.Node - var err error - if req.Shard == ^uint64(0) { - if qcx != nil && c.api != nil { - err = c.api.ImportWithTx(ctx, qcx, req, options) - // Note that Wrap(nil, ...) is still nil. - return errors.Wrap(err, "local import") + var process func() error + if c.api != nil && qcx != nil { + process = func() error { + return c.api.ImportWithTx(ctx, qcx, req, options) } } - nodes, err = c.FragmentNodes(ctx, req.Index, req.Shard) - if err != nil { - return errors.Errorf("shard nodes: %s", err) - } - // "us" is a usable local node if any, "them" is every node that we need - // to process which isn't that node. We start out with us == nil and - // them = the whole set of nodes. - var us *topology.Node - var them []*topology.Node = nodes - - // If we have an API, we know what node we are. Even if we don't have - // a Qcx, we still care, because looping back to the local node will - // be faster than going to another node. - if c.api != nil { - myID := c.api.NodeID() - for i, node := range nodes { - if myID == node.ID { - // swap our node into the first position - nodes[i], nodes[0] = nodes[0], nodes[i] - // If we have a qcx, we'll treat our node even MORE - // specially. - if qcx != nil { - us, them = nodes[0], nodes[1:] - } - break - } - } - } - - // If we had a valid API and Qcx, and shard was ^0, we'd have handled - // it previously. So if we get here, we don't have both a Qcx and an API, - // but we might have an API, in which case we'll have shuffled our node - // into the first position. Otherwise we're just taking whatever the first - // node is. - if req.Shard == ^uint64(0) { - them = them[:1] - } - - // We handle remote nodes first, for two distinct reasons. One is that - // the local API ImportWithTx is allowed to modify its inputs, and if we - // ran that before serializing, we'd get corrupt data serialized. - // The other is that if we were to hold a write lock that started with - // that import and ended when we hit the end of our Qcx, we wouldn't want - // to hold it during all our requests to the remote nodes. - if len(them) > 0 { - buf, err := c.serializer.Marshal(req) - if err != nil { - return errors.Errorf("marshal import request: %s", err) - } - // We process remote nodes first so we won't be actually holding our - // write lock yet, in theory. This doesn't actually matter yet, but is - // helpful for future planned refactoring. - for _, node := range them { - if err = c.importNode(ctx, node, req.Index, req.Field, buf, options); err != nil { - return errors.Wrap(err, "remote import") - } - } - } - - // Write to the local node if we have one. - if us != nil { - // WARNING: ImportWithTx can alter its inputs. However, we can - // only ever do this once, and if we're going to need a marshalled - // form, we already made it. - if err = c.api.ImportWithTx(ctx, qcx, req, options); err != nil { - return errors.Wrap(err, "local import") - } - } - return nil + return c.importHelper(ctx, req, process, req.Index, req.Field, req.Shard, options) } // ImportValue imports values using an ImportValueRequest, whether or not it's @@ -749,94 +777,19 @@ func (c *InternalClient) Import(ctx context.Context, qcx *pilosa.Qcx, req *pilos // If we get a non-nil qcx, and have an associated API, we'll use that API // directly for the local shard. func (c *InternalClient) ImportValue(ctx context.Context, qcx *pilosa.Qcx, req *pilosa.ImportValueRequest, options *pilosa.ImportOptions) error { - span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportValue") + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import") defer span.Finish() if req.ColumnKeys != nil { req.Shard = ^uint64(0) } - // If we don't actually know what shards we're sending to, and we have - // a local API, we can short-circuit: We just send this request to the - // local API, which will break it out and call its defaultClient for - // the data for individual shards. - var nodes []*topology.Node - var err error - if req.Shard == ^uint64(0) { - if qcx != nil && c.api != nil { - err = c.api.ImportValueWithTx(ctx, qcx, req, options) - // Note that Wrap(nil, ...) is still nil. - return errors.Wrap(err, "local import") + var process func() error + if c.api != nil && qcx != nil { + process = func() error { + return c.api.ImportValueWithTx(ctx, qcx, req, options) } } - nodes, err = c.FragmentNodes(ctx, req.Index, req.Shard) - if err != nil { - return errors.Errorf("shard nodes: %s", err) - } - // "us" is a usable local node if any, "them" is every node that we need - // to process which isn't that node. We start out with us == nil and - // them = the whole set of nodes. - var us *topology.Node - var them []*topology.Node = nodes - - // If we have an API, we know what node we are. Even if we don't have - // a Qcx, we still care, because looping back to the local node will - // be faster than going to another node. - if c.api != nil { - myID := c.api.NodeID() - for i, node := range nodes { - if myID == node.ID { - // swap our node into the first position - nodes[i], nodes[0] = nodes[0], nodes[i] - // If we have a qcx, we'll treat our node even MORE - // specially. - if qcx != nil { - us, them = nodes[0], nodes[1:] - } - break - } - } - } - - // If we had a valid API and Qcx, and shard was ^0, we'd have handled - // it previously. So if we get here, we don't have both a Qcx and an API, - // but we might have an API, in which case we'll have shuffled our node - // into the first position. Otherwise we're just taking whatever the first - // node is. - if req.Shard == ^uint64(0) { - them = them[:1] - } - - // We handle remote nodes first, for two distinct reasons. One is that - // the local API ImportWithTx is allowed to modify its inputs, and if we - // ran that before serializing, we'd get corrupt data serialized. - // The other is that if we were to hold a write lock that started with - // that import and ended when we hit the end of our Qcx, we wouldn't want - // to hold it during all our requests to the remote nodes. - if len(them) > 0 { - buf, err := c.serializer.Marshal(req) - if err != nil { - return errors.Errorf("marshal import request: %s", err) - } - // We process remote nodes first so we won't be actually holding our - // write lock yet, in theory. This doesn't actually matter yet, but is - // helpful for future planned refactoring. - for _, node := range them { - if err = c.importNode(ctx, node, req.Index, req.Field, buf, options); err != nil { - return errors.Wrap(err, "remote import") - } - } - } - - // Write to the local node if we have one. - if us != nil { - // WARNING: ImportWithTx can alter its inputs. However, we can - // only ever do this once, and if we're going to need a marshalled - // form, we already made it. - if err = c.api.ImportValueWithTx(ctx, qcx, req, options); err != nil { - return errors.Wrap(err, "local import") - } - } - return nil + return c.importHelper(ctx, req, process, req.Index, req.Field, req.Shard, options) } // ImportRoaring does fast import of raw bits in roaring format (pilosa or