(*
  Copyright 2012 Gerd Stolpmann

  This file is part of Plasma, a distributed filesystem and a
  map/reduce computation framework. Unless you have a written license
  agreement with the copyright holder (Gerd Stolpmann), the following
  terms apply:

  Plasma is free software: you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published by
  the Free Software Foundation, either version 3 of the License, or
  (at your option) any later version.

  Plasma is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with Plasma.  If not, see <http://www.gnu.org/licenses/>.

*)
(* $Id: mapred_rfun.ml 561 2012-01-06 00:34:10Z gerd $ *)

open Printf

type 't rfun =
  | Rfun of string
  | Apply of Obj.t * (Obj.t->Obj.t) rfun
  constraint 't = _ -> _ 

exception Rfun_not_found of string

let tab = (Hashtbl.create 19 : (string, Obj.t -> Obj.t) Hashtbl.t) ;;
let finished = ref false
let p4_file = ref None
let p4_prefix = ref None

let register name f =
  if !finished then 
    failwith "Mapred_rfun: it is too late for registering a function";
  if Hashtbl.mem tab name then
    failwith "Mapred_rfun: the name is already used";
  Hashtbl.add tab name (Obj.magic f);
  Rfun name

let finish() =
  finished := true

let rec lookup_i rf =
  match rf with
    | Rfun name ->
	( try
	    let f = Hashtbl.find tab name  in
	    f
	  with
	    | Not_found -> raise(Rfun_not_found name)
	)
    | Apply(x,rf') ->
	let f = lookup_i rf' in
	(Obj.magic f : _ -> _) x

let lookup =
  ( fun rf ->
      Obj.magic (lookup_i (Obj.magic rf))
	: ('a -> 'b) rfun -> 'a -> 'b
  )

let rec get_id_i rf =
  match rf with
    | Rfun id -> id
    | Apply(_,rf) -> get_id_i rf

let get_id =
  ( fun rf ->
      get_id_i (Obj.magic rf)
	: ('a -> 'b) rfun -> string
  )

let apply_partially rf x =
  Apply(Obj.magic x, Obj.magic rf)

let p4_new_file get_loc =
  let file = 
    try get_loc(); failwith "Mapred_rfun.p4_new_file"
    with 
      | Assert_failure(file,_,_) -> file in
  p4_file := Some file;
  p4_prefix := Some(Digest.to_hex(Digest.string (file (* ^ "/" ^ 
						   string_of_float
						    (Unix.gettimeofday()) *)  )))

let p4_register get_loc f =
  if !finished then 
    failwith "Mapred_rfun: it is too late for registering a function";
  let (file, line, col) = 
    try get_loc(); failwith "Mapred_rfun.p4_new_file"
    with 
      | Assert_failure(file,line,col) -> (file,line,col) in
  if !p4_file <> Some file then
    failwith "Mapred_rfun.p4_register";
  let prefix =
    match !p4_prefix with
      | None -> assert false
      | Some p -> p in
  register (sprintf "%s_%d_%d" prefix line col) f