Handle racing conditions for common meta-data

[vkuznet]

The bulkblocks API internally has racing conditions, see full discussion dmwm/WMCore#11106

To resolve this we can use Queue'ing system to insert common meta-data:

• Create independent Queues for individual DBS database tables, e.g. ProcessedDatasetQueue
• adjust bulkblocks API to use appropriate queue for given metadata, e.g. adjust getProcessedDatasetID to use ProcessedDatasetQueue instead of GetRecID call, see line-354
• each individual queue will process requests sequentially, e.g. FIFO. This will eliminate racing condition among independent and competing HTTP requests since the specific common meta-data will be either injected or obtained its ID from its queue.

We need:

• new queueProcess injector which will consume requests for injection and process them in FIFO order
• blocking channel which will be used before we inject the data

For instance,

type QueueDataRecord struct {
  Record dbs.DBRecord
  Out chan bool
}

rec // represent data record we received from bulkblocks API
// process engine to process records in Queue
go processQueue()

// channel will block us until all records are injected
ch := make(chan bool)

// send data record to the queue
queue <- QueueDataRecord{Record: rec, Out: ch}

// consume data when it is ready
for {
   select {
   case r := <- ch
        // call GetIDs of common meta-data
   }
}

and processQueue function will do something like

func processQueue() {
   for {
      select {
         case rec := <- queue {
             // process given record
             // send out its status that it is ready
             rec.Out <- true
         default:
             time.Sleep(time.Duration(1) * time.Second) // wait for new task
      }
   }
}

[vkuznet]

Further thoughts on this:

• this approach will not work if we'll have multiple independent DBSWriters (k8s pods) since they do not share the queue
• therefore, in order to make it work we either need to have persistent queue storage (among all k8s pods) or have individual ONE additional server which will process these records sequentially