Class: Batchwatch::Tracking

Inherits:
Object
  • Object
show all
Defined in:
lib/batchwatch/client.rb

Overview

Handle for one measurement in flight. Returned by track().

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(bw, provider, model, mode, requests, input_tokens, endpoint) ⇒ Tracking

Returns a new instance of Tracking.



532
533
534
535
536
537
538
539
540
541
542
543
 
# File 'lib/batchwatch/client.rb', line 532

def initialize(bw, provider, model, mode, requests, input_tokens, endpoint)
  @bw = bw
  @body = { "provider" => provider, "model" => model, "mode" => mode,
            "requests" => requests, "endpoint" => endpoint }
  @input_tokens = input_tokens
  @id = nil
  @t0 = Time.now
  @finished = false
  @lock = Mutex.new
  @id_ready = false
  @id_cond = ConditionVariable.new
end

Instance Attribute Details

#id ⇒ Object (readonly)

Returns the value of attribute id.



530
531
532
 
# File 'lib/batchwatch/client.rb', line 530

def id
  @id
end

Instance Method Details

#done(output_tokens: nil, status: "completed", ttfb_ms: nil) ⇒ Object

Close the measurement.

output_tokens stays nil when you do not know it. It is never defaulted to zero: zero is a measurement, absence is not, and the server prices them differently on purpose.



585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
 
# File 'lib/batchwatch/client.rb', line 585

def done(output_tokens: nil, status: "completed", ttfb_ms: nil)
  @lock.synchronize do
    return if @finished

    @finished = true
  end
  ended = Time.now

  # The outcome measurement (#101): the latest advice for THIS model,
  # attached to the completion. No advice -> empty hash -> the fields are
  # omitted entirely.
  advice = Batchwatch.outcome_fields(@bw.get_advice(@body["model"]))

  @bw.in_background do
    # Wait briefly for the id to land from the start call.
    wait_for_id
    id = @lock.synchronize { @id }

    full = @body.dup
    full.merge!("input_tokens" => @input_tokens,
                "output_tokens" => output_tokens, "status" => status,
                "started_at" => Batchwatch.iso(@t0),
                "ended_at" => Batchwatch.iso(ended))
    full["ttfb_ms"] = ttfb_ms unless ttfb_ms.nil?
    full.merge!(advice)

    if id
      begin
        @bw.call("/v1/calls/#{id}", method: "PATCH", body: Batchwatch.sanitize(
          { "status" => status, "output_tokens" => output_tokens,
            "ttfb_ms" => ttfb_ms, "ended_at" => Batchwatch.iso(ended) }.merge(advice)
        ))
        next
      rescue StandardError
        # The server already has the start. A spooled resend can therefore
        # produce a duplicate if the PATCH reached the server anyway -
        # chosen on purpose: a duplicate can be seen in the dataset, a lost
        # measurement cannot.
        @bw.spool_measurement(Batchwatch.sanitize(full))
        next
      end
    end

    # The start call never reached the server. Send the whole measurement
    # in one go instead of losing it. The key is derived from the
    # measurement, so a later spool replay of exactly this record carries
    # the SAME key and is deduped (#30).
    cleaned = Batchwatch.sanitize(full)
    begin
      @bw.call("/v1/calls/complete", method: "POST", body: cleaned,
                                     idem: Batchwatch.idem_complete([cleaned]))
    rescue StandardError
      @bw.spool_measurement(cleaned)
    end
  end
end

#failed(status: "failed") ⇒ Object

Record the job as not-completed. An unfinished wait is not a wait.



643
644
645
 
# File 'lib/batchwatch/client.rb', line 643

def failed(status: "failed")
  done(status: status)
end

#finished? ⇒ Boolean

Returns:

  • (Boolean) 


545
546
547
 
# File 'lib/batchwatch/client.rb', line 545

def finished?
  @lock.synchronize { @finished }
end

#start ⇒ Object 



549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
 
# File 'lib/batchwatch/client.rb', line 549

def start
  body = @body.dup
  body["input_tokens"] = @input_tokens
  body["started_at"] = Batchwatch.iso(@t0)

  start_body = Batchwatch.sanitize(body)
  @bw.in_background do
    begin
      r = @bw.call("/v1/calls", method: "POST", body: start_body,
                                idem: Batchwatch.idem_start(start_body))
      @lock.synchronize do
        @id = r && r["id"]
        @id_ready = true
        @id_cond.broadcast
      end
      @bw.maybe_flush_spool if r
    rescue StandardError
      @lock.synchronize do
        @id_ready = true
        @id_cond.broadcast
      end
      raise
    end
  end
end

#started(input_tokens: nil) ⇒ Object

Update the token count when it is only known after submission.



576
577
578
 
# File 'lib/batchwatch/client.rb', line 576

def started(input_tokens: nil)
  @input_tokens = input_tokens unless input_tokens.nil?
end