protocol MpegAudioVideoAdaptation -- mpeg packet format is |udp|ver|seqno|p_type|frame|payload len|payload -- This policy is designed for astero1.mpg with 400Kbps bandwidth requirement -- frame pattern is IBBPBBPBBPBBPBB initstate ( -- map (movie index + interface) to drop information mkTable(25), -- map a movie IP address to an index mkTable(25)) is val I_frame : int = 1 val P_frame : int = 2 val B_frame : int = 3 fun update_table(interface_table : (int * int) hash_table, i : int, pictype : int, frame_no : int) : unit = insert(interface_table, hashKey(i), (pictype,frame_no)) fun available_bandwidth(i : interface) : int = 2000000 - InterfaceLoad(i) fun B_forward(interface_table : (int * int) hash_table, threshold : int, index : int, i : interface, p : (ip * udp * int * int * int * int * int * blob)) : unit = try let val dropped : (int * int) = find(interface_table, hashKey(index)) in let val undecodable : bool = (((#1 dropped) = P_frame) or ((#2 dropped) = (#6 p))) in if (not undecodable) then if (available_bandwidth(i) < threshold) then update_table(interface_table,index,B_frame,#6 p) else (OnNeighbor(network, i, p); ()) else () end end handle NoEntry => () -- no client for this movie at the interface fun P_forward(interface_table : (int * int) hash_table, threshold : int, index : int, i : interface, p : (ip * udp * int * int * int * int * int * blob)) : unit = try let val dropped : (int * int) = find(interface_table, hashKey(index)) in let val undecodable : bool = ((#1 dropped) = P_frame) in if (not undecodable) then if (available_bandwidth(i) < threshold) then update_table(interface_table,index,P_frame,#6 p) else (OnNeighbor(network, i, p); ()) else () end end handle NoEntry => () -- no client for this movie at the interface fun I_forward(interface_table : (int * int) hash_table, index : int, i : interface, p : (ip * udp * int * int * int * int * int * blob)) : unit = try (find(interface_table, hashKey(index)); OnNeighbor(network, i, p); update_table(interface_table,index,0,0)) handle NoEntry => () -- We would like a hash table that -- maps a tuple of a movie and an interface to dropped information. -- But a hash key can't be a tuple. So we convert the movie address to -- an integer that is a multiple of 10, convert the interface to a number -- (hoping that getInterfaces() always returns the interfaces in the same -- order and that there are fewer than 10 of them) and then add the -- movie number to the interface number to get an index. fun adaptiveForward (mp : (int * int * int) * (ip * udp * int * int * int * int * int * blob) * ((int * int) hash_table), i : interface) : (int * int * int) * (ip * udp * int * int * int * int * int * blob) * ((int * int) hash_table) = let val m : (int * int * int) = #1 mp val p : (ip * udp * int * int * int * int * int * blob) = #2 mp val interface_table : (int * int) hash_table = #3 mp in let val index : int = #1 m val P_threshold : int = #2 m val B_threshold : int = #3 m val pictype : int = #5 p in (if (not (i = thisInterface())) then if (pictype = B_frame) then B_forward(interface_table,B_threshold,index,i,p) else if (pictype = P_frame) then P_forward(interface_table,P_threshold,index,i,p) else I_forward(interface_table,index,i,p) else (); ((index+1, P_threshold, B_threshold), p, interface_table)) end end -- int^5 are defined as {ver, seqno, frame_type, frame, size} stream network(ps : (((int * int) hash_table) * ((int * int * int) hash_table) ), ss : unit, p : (ip * udp * int * int * int * int * int * blob)) is try let val m : (int * int * int) = find(#2 ps,hashkey(ipDst(#1 p))) in (foldl(adaptiveForward,(m,p,#1 ps),getInterfaces()); (ps,ss,1)) end handle NoEntry => (OnRemote(network, p); (ps,ss,2)) -- P_threshold and B_threshold should be calculated from the bandwidth -- requirement val Add : int = 0 val Delete : int = 1 fun interfaceToIntLoop (x : (interface * int * int), i : interface) : (interface * int * int) = if ((#1 x) = i) then (#1 x, #2 x, #2 x) else x fun interfaceToInt(x :interface) : int = #3 (foldl(interfaceToIntLoop,(x,0,0),getInterfaces())) fun remove_one(im : ((int * int) hash_table) * int, i : interface) : ((int * int) hash_table) * int = let val interface_table : ((int * int) hash_table) = #1 im val movie_index : int = #2 im in (remove(interface_table,hashKey(movie_index)); (interface_table, movie_index + 1)) end -- the following addresses reflect the astero2 data -- 225.3.2.1: multicast address of the movie -- 147.210.18.1: address of the server -- 172.17.1.1: address of the router -- below 225.3.2.1 should be p->movie_address, to be more realistic stream network(ps : (((int * int) hash_table) * ((int * int * int) hash_table)), ss : int, p : (ip * tcp * host * blob)) initstate 0 is let val interface_table : (int * int) hash_table = #1 ps val movie_table : (int * int * int) hash_table = #2 ps val max_movie : int = ss val dest : host = ipDst(#1 p) in -- address of the server: subscription request if (dest = 147.210.18.1) then try let val m : (int * int * int) = find(movie_table, hashKey(225.3.2.1)) in let val movie_index : int = #1 m -- just one client val i : interface = thisInterface() in let val index : int = movie_index + interfaceToInt(i) in -- if the following hash table lookup fails, we have the first -- client for the movie at this interface try (find(interface_table, hashKey(index)); (ps,ss,2)) handle NoEntry => (update_table(interface_table,index,0,0); (ps,ss,2)) end end end handle NoEntry => (print("error: unknown movie: "); println(225.3.2.1); (ps,ss,2)) -- address of the router: movie add or delete request else if (dest = 172.17.1.1) then let val m : (int * int * int * blob) = (#4 p) in if ((#1 m) = Delete) then try let val m : (int * int * int) = find(movie_table, hashKey(225.3.2.1)) in let val movie_index : int = (#1 m) in (foldl(remove_one,(interface_table,movie_index), getInterfaces()); remove(movie_table, hashKey(225.3.2.1)); (ps,ss,2)) end end handle NoEntry => (ps,ss,2) else -- ((#1 m) = Add) try (find(movie_table, hashKey(225.3.2.1)); (ps,ss,2)) handle NoEntry => -- 100000 should be m->P_threshold -- 300000 should be m->B_threshold (insert(movie_table,hashKey(225.3.2.1), (max_movie*10,100000,300000)); ((interface_table, movie_table), max_movie + 1, 2)) end else (OnRemote(network, p); (ps,ss,2)) end