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.



692
693
694
695
696
697
698
699
700
701
702
703
 
# File 'lib/batchwatch/client.rb', line 692

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.



690
691
692
 
# File 'lib/batchwatch/client.rb', line 690

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.



745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
 
# File 'lib/batchwatch/client.rb', line 745

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.



803
804
805
 
# File 'lib/batchwatch/client.rb', line 803

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

#finished? ⇒ Boolean

Returns:

  • (Boolean) 


705
706
707
 
# File 'lib/batchwatch/client.rb', line 705

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

#start ⇒ Object 



709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
 
# File 'lib/batchwatch/client.rb', line 709

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.



736
737
738
 
# File 'lib/batchwatch/client.rb', line 736

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