code: func (c *CassandraStore) processWriteQueue(queue chan *mdata.ChunkWriteRequest, meter *stats.Range32) {
  	tick := time.Tick(time.Duration(1) * time.Second)
  	for {
  		select {
  		case <-tick:
  			meter.Value(len(queue))
  		case cwr := <-queue:
  			meter.Value(len(queue))
  			log.Debugf("CS: starting to save %s:%d %v", cwr.Key, cwr.Chunk.Series.T0, cwr.Chunk)
  			//log how long the chunk waited in the queue before we attempted to save to cassandra
  			cassPutWaitDuration.Value(time.Now().Sub(cwr.Timestamp))
  
  			buf := cwr.Chunk.Encode(cwr.Span)
  			chunkSizeAtSave.Value(len(buf))
  			success := false
  			attempts := 0
  			keyStr := cwr.Key.String()
  			for !success {
  				err := c.insertChunk(keyStr, cwr.Chunk.Series.T0, cwr.TTL, buf)
  
  				if err == nil {
  					success = true
  					cwr.Metric.SyncChunkSaveState(cwr.Chunk.Series.T0)
  					mdata.SendPersistMessage(keyStr, cwr.Chunk.Series.T0)
  					log.Debugf("CS: save complete. %s:%d %v", keyStr, cwr.Chunk.Series.T0, cwr.Chunk)
  					chunkSaveOk.Inc()
  				} else {
  					errmetrics.Inc(err)
  					if (attempts % 20) == 0 {
  						log.Warnf("CS: failed to save chunk to cassandra after %d attempts. %v, %s", attempts+1, cwr.Chunk, err)
  					}
  					chunkSaveFail.Inc()
  					sleepTime := 100 * attempts
  					if sleepTime > 2000 {
  						sleepTime = 2000
  					}
  					time.Sleep(time.Duration(sleepTime) * time.Millisecond)
  					attempts++
  				}
  			}
  		}
  	}
  }
docs: /* process writeQueue.
   */