/ - Diff - PeerStreamer-ng - Redmine

     struct chunk_interval_estimate {
     	uint32_t first_index;
     	uint32_t last_index;
     	uint64_t first_timestamp;
     	suseconds_t chunk_interval;
     	enum data_state state;
-...
     		switch (m->cie.state) {
     			case deinit:
     				m->cie.first_index = cid;
     				m->cie.last_index = cid;
     				m->cie.first_timestamp = ctimestamp;
     				m->cie.state = loading;
     				break;
     			case loading:
     			case ready:
     				if (cid > (int64_t)m->cie.first_index)
     				if (cid > (int64_t)m->cie.last_index)
+    				{
     					m->cie.chunk_interval = ((ctimestamp - m->cie.first_timestamp))/(cid - m->cie.first_index);
     					m->cie.last_index = cid;
+    				}
     				m->cie.state = ready;
     				break;
+    		}

     /*
      * Copyright (c) 2017 Luca Baldesi
+     *
      * This file is part of PeerStreamer.
+     *
      * PeerStreamer is free software: you can redistribute it and/or
      * modify it under the terms of the GNU Affero General Public License as
      * published by the Free Software Foundation, either version 3 of the
      * License, or (at your option) any later version.
+     *
      * PeerStreamer 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 Affero
      * General Public License for more details.
+     *
      * You should have received a copy of the GNU Affero General Public License
      * along with PeerStreamer.  If not, see <http://www.gnu.org/licenses/>.
+     *
      */
     #include<peer_metadata.h>
     #include<malloc.h>
     #include<string.h>

     /*
      * Copyright (c) 2017 Luca Baldesi
+     *
      * This file is part of PeerStreamer.
+     *
      * PeerStreamer is free software: you can redistribute it and/or
      * modify it under the terms of the GNU Affero General Public License as
      * published by the Free Software Foundation, either version 3 of the
      * License, or (at your option) any later version.
+     *
      * PeerStreamer 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 Affero
      * General Public License for more details.
+     *
      * You should have received a copy of the GNU Affero General Public License
      * along with PeerStreamer.  If not, see <http://www.gnu.org/licenses/>.
+     *
      */
     #ifndef __PEER_METADATA_H__
     #define __PEER_METADATA_H__

     				else
     					received_chunk(ps->streaming, remote, buff, len);
     				res = 2;
     				ps->chunk_offer_interval = chunk_interval_measure(ps->measure);
     				break;
     			case MSG_TYPE_SIGNALLING:
     				dtprintf("Sign message received:\n");
-...
     				fprintf(stderr, "Unknown Message Type %x\n", buff[0]);
     				res = -2;
+    		}
     		ps->chunk_offer_interval = streaming_offer_interval(ps->streaming);
     	if (remote)
     		nodeid_free(remote);

     #include "peer_metadata.h"
     #include "scheduler_la.h"
     #include "grapes_config.h"
     # define CB_SIZE_TIME_UNLIMITED 1e12
-...
       uint16_t hopcount;
     } __attribute__((packed));
     enum distribution_type {DIST_UNIFORM, DIST_TURBO};
     struct streaming_context {
     	struct chunk_buffer *cb;
     	struct input_desc *input;
-...
     	bool send_bmap_before_push;
     	uint64_t CB_SIZE_TIME;
     	uint32_t chunk_loss_interval;
     	enum distribution_type dist_type;
     	int offer_per_period;
     };
     struct streaming_context * streaming_create(const struct psinstance * ps, struct input_context * inc, const char * config)
+    {
     	struct streaming_context * stc;
     	struct tag * tags;
     	static char conf[80];
     	stc = malloc(sizeof(struct streaming_context));
     	stc->ps = ps;
     	stc->bcast_after_receive_every = 0;
     	stc->neigh_on_chunk_recv = false;
     	stc->send_bmap_before_push = false;
     	stc->transactions = NULL;
     	stc->CB_SIZE_TIME = CB_SIZE_TIME_UNLIMITED;	//in millisec, defaults to unlimited
     	stc->chunk_loss_interval = 0;  // disable self-lossy feature (for experiments)
     	stc->dist_type = DIST_UNIFORM;
     	stc->input = inc ? input_open(inc->filename, inc->fds, inc->fds_size, config) : NULL;
     	tags = grapes_config_parse(config);
     	if (strcmp(grapes_config_value_str_default(tags, "dist_type", ""), "turbo") == 0)
     		stc->dist_type = DIST_TURBO;
     	grapes_config_value_int_default(tags, "offer_per_period", &(stc->offer_per_period), 1);
     	free(tags);
     	stc->ps = ps;
     	stc->input = inc ? input_open(inc->filename, inc->fds, inc->fds_size, config) : NULL;
     	stc->cb_size = psinstance_chunkbuffer_size(ps);
     	sprintf(conf, "size=%d", stc->cb_size);
     	stc->cb = cb_init(conf);
     	stc->ch_locks = chunk_locks_create();
     	chunkDeliveryInit(psinstance_nodeid(ps));
     	chunkSignalingInit(psinstance_nodeid(ps));
     	stc->ch_locks = chunk_locks_create();
     	return stc;
+    }
-...
       return 1;
+    }
     double peerWeightNeigh(struct peer **n){
     	double res;
     	res = (double)(((struct metadata *)(*n)->metadata)->neigh_size);
     	if (res)
     		return res;
     	return 1;
+    }
     double peerWeightInvNeigh(struct peer **n){
     	double res;
     	res = peerWeightNeigh(n);
     	return 1/res;
+    }
     double peerWeightLoss(struct peer **n){
       return 1;
+    }
-...
         for (i = 0;i < size; i++) chunkids[size - 1 - i] = (buff+i)->id;
         for (i = 0; i<n; i++) nodeids[i] = neighbours[i];
         selectPeersForChunks(SCHED_WEIGHTING, nodeids, n, chunkids, size, selectedpeers, &selectedpeers_len, SCHED_NEEDS, SCHED_PEER);
         if (stc->dist_type == DIST_TURBO)
     	    selectPeersForChunks(SCHED_WEIGHTING, nodeids, n, chunkids, size, selectedpeers, &selectedpeers_len, SCHED_NEEDS, peerWeightInvNeigh);
         else
     	    selectPeersForChunks(SCHED_WEIGHTING, nodeids, n, chunkids, size, selectedpeers, &selectedpeers_len, SCHED_NEEDS, peerWeightUniform);
         for (i=0; i<selectedpeers_len ; i++){
           int transid = transaction_create(stc->transactions, selectedpeers[i]->id);
-...
       //SCHED_TYPE(SCHED_WEIGHTING, peers, peers_num, &(target_chunk->id), 1, selectedpairs, &selectedpairs_len, SCHED_NEEDS, peer_evaluation, SCHED_CHUNK);
      	dst_peers = (struct peer **) malloc(sizeof(struct peer* ) *  multiplicity);
      	selectPeersForChunks(SCHED_WEIGHTING, peers, peers_num, (int *)&(target_chunk->id) , 1, dst_peers, &selectedpairs_len, NULL, peer_evaluation);
     	if (stc->dist_type == DIST_TURBO)
     		selectPeersForChunks(SCHED_WEIGHTING, peers, peers_num, (int *)&(target_chunk->id) , 1, dst_peers, &selectedpairs_len, NULL, peerWeightNeigh);
     	else
     		selectPeersForChunks(SCHED_WEIGHTING, peers, peers_num, (int *)&(target_chunk->id) , 1, dst_peers, &selectedpairs_len, NULL, peer_evaluation);
     	selectedpairs = (struct PeerChunk *)  malloc(sizeof(struct PeerChunk) * selectedpairs_len);
     	for ( i=0; i<selectedpairs_len; i++)
-...
+        }
+      }
+    }
     suseconds_t streaming_offer_interval(const struct streaming_context *stc)
+    {
     	struct peerset *pset;
     	const struct peer * p;
     	int i;
     	double load = 0;
     	suseconds_t offer_int;
     	offer_int = chunk_interval_measure(psinstance_measures(stc->ps));
     	if (stc->dist_type == DIST_TURBO) {
     		pset = topology_get_neighbours(psinstance_topology(stc->ps));
     		peerset_for_each(pset,p,i)
     			load += peerWeightInvNeigh((struct peer **)&p);
     		offer_int /= load;
+    	}
     	return offer_int / stc->offer_per_period;
+    }

src/streaming.h
60	60
61	61	void streaming_destroy(struct streaming_context ** stc);
62	62
	63	suseconds_t streaming_offer_interval(const struct streaming_context *stc);
	64
63	65	#endif /* STREAMING_H */

     	if(name) fprintf(stderr,"%s\n",name);
     	if(pset)
     		peerset_for_each(pset,p,i)
     			fprintf(stderr, "\t%s\n", nodeid_static_str(p->id));
     			fprintf(stderr, "\t%s, cbsize: %d, neighsize: %d\n", nodeid_static_str(p->id), peer_cb_size(p), peer_neigh_size(p));
+    }
     void update_metadata(struct topology * t)
-...
     	metadata_update(&(t->my_metadata),
     			psinstance_is_source(t->ps) ? 0 : psinstance_chunkbuffer_size(t->ps),
     			peerset_size(t->neighbourhood));
     	psample_change_metadata(t->tc, &(t->my_metadata), sizeof(struct metadata));
+    }
     struct peer * topology_get_peer(struct topology * t, const struct nodeID * id)
-...
     	t->topo_in = true; //peer selects in-neighbours (combined means bidirectional)
     	grapes_config_value_int_default(tags, "neighbourhood_size", &(t->neighbourhood_target_size), DEFAULT_PEER_NEIGH_SIZE);
     	update_metadata(t);
     	t->tc = psample_init(psinstance_nodeid(ps), &(t->my_metadata), sizeof(struct metadata), config);
     	update_metadata(t);
     	free(tags);
     	return t->tc && t->neighbourhood && t->swarm_bucket ? 1 : 0;
-...
     			peerset_add_peer(t->swarm_bucket,sample_nodes[i]);
     			p = topology_get_peer(t, sample_nodes[i]);
+    		}
     		else
     			//fprintf(stderr,"[DEBUG] OLD PEER!\n");
     		peer_set_metadata(p,&(sample_metas[i]));
     		fprintf(stderr,"[DEBUG] sampled node: %s, cbsize: %d, neighsize: %d\n",nodeid_static_str(sample_nodes[i]), sample_metas[i].cb_size, sample_metas[i].neigh_size);
+    	}
+    }
-...
         if ( (!timerisset(&peers[i]->bmap_timestamp) && timercmp(&peers[i]->creation_timestamp, &told, <) ) ||
              ( timerisset(&peers[i]->bmap_timestamp) && timercmp(&peers[i]->bmap_timestamp, &told, <)     )   ) {
           dprintf("Topo: dropping inactive %s (peersset_size: %d)\n", nodeid_static_str(peers[i]->id), peerset_size(t->neighbourhood));
           //if (peerset_size(t->neighbourhood) > 1) {	// avoid dropping our last link to the world
           topology_blacklist_add(t, peers[i]->id);
           neighbourhood_remove_peer(t, peers[i]->id);
           //}
           if (peerset_size(t->neighbourhood) > 1) {	// avoid dropping our last link to the world
     	      topology_blacklist_add(t, peers[i]->id);
     	      neighbourhood_remove_peer(t, peers[i]->id);
+          }
+        }
+      }
-...
     void topology_update(struct topology * t)
+    {
     	struct peerset * old_neighs;
       const struct peer * p;
       int i;
     	const struct peer * p;
     	int i;
       psample_parse_data(t->tc,NULL,0); // needed in order to trigger timed sending of TOPO messages
     	psample_parse_data(t->tc,NULL,0); // needed in order to trigger timed sending of TOPO messages
     	update_metadata(t);
     	topology_sample_peers(t);
       if timerisset(&(t->tout_bmap) )
     	if timerisset(&(t->tout_bmap) )
     		neighbourhood_drop_unactives(t, &(t->tout_bmap));
     	old_neighs = peerset_create_reference_copy(t->neighbourhood);
         topology_update_random(t);
     	topology_update_random(t);
       topology_signal_change(t, old_neighs);
     	topology_signal_change(t, old_neighs);
     	peerset_destroy_reference_copy(&old_neighs);
         peerset_for_each(t->swarm_bucket,p,i)
           peerset_pop_peer(t->locked_neighs,p->id);
         peerset_clear(t->swarm_bucket,0);  // we don't remember past peers
     	peerset_for_each(t->swarm_bucket,p,i)
     		peerset_pop_peer(t->locked_neighs,p->id);
     	peerset_clear(t->swarm_bucket,0);  // we don't remember past peers
+    }
     void topology_destroy(struct topology **t)

PeerStreamer-ng

Revision dc99f5c4