summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorerdgeist <>2007-11-21 01:55:15 +0000
committererdgeist <>2007-11-21 01:55:15 +0000
commit4072f162b426d2e250d4b95d55e81b6908b8b509 (patch)
tree4befc6e9e80068b1943c519570277e00b6f78df9
parente65a41d0006d463c7062124fc1a2c7a471200a1b (diff)
Make sync generation multithreaded.
-rw-r--r--ot_sync.c121
1 files changed, 90 insertions, 31 deletions
diff --git a/ot_sync.c b/ot_sync.c
index 6e95a98..5261ee6 100644
--- a/ot_sync.c
+++ b/ot_sync.c
@@ -6,6 +6,7 @@
6#include <sys/mman.h> 6#include <sys/mman.h>
7#include <stdio.h> 7#include <stdio.h>
8#include <string.h> 8#include <string.h>
9#include <pthread.h>
9 10
10/* Libowfat */ 11/* Libowfat */
11#include "scan.h" 12#include "scan.h"
@@ -17,6 +18,9 @@
17#include "ot_sync.h" 18#include "ot_sync.h"
18 19
19#ifdef WANT_TRACKER_SYNC 20#ifdef WANT_TRACKER_SYNC
21
22#define OT_SYNC_CHUNK_SIZE (512*1024)
23
20/* Import Changeset from an external authority 24/* Import Changeset from an external authority
21 format: d4:syncd[..]ee 25 format: d4:syncd[..]ee
22 [..]: ( 20:01234567890abcdefghij16:XXXXYYYY )+ 26 [..]: ( 20:01234567890abcdefghij16:XXXXYYYY )+
@@ -60,48 +64,103 @@ int add_changeset_to_tracker( ot_byte *data, size_t len ) {
60/* Proposed output format 64/* Proposed output format
61 d4:syncd20:<info_hash>8*N:(xxxxyyyy)*Nee 65 d4:syncd20:<info_hash>8*N:(xxxxyyyy)*Nee
62*/ 66*/
63size_t return_changeset_for_tracker( char **reply ) { 67static void sync_make( int *iovec_entries, struct iovec **iovector ) {
64 size_t allocated = 0, i, replysize;
65 ot_vector *torrents_list;
66 int bucket; 68 int bucket;
67 char *r; 69 char *r, *re;
68 70
69 /* Maybe there is time to clean_all_torrents(); */ 71 /* Setup return vector... */
72 *iovec_entries = 0;
73 *iovector = NULL;
74 if( !( r = iovec_increase( iovec_entries, iovector, OT_SYNC_CHUNK_SIZE ) ) )
75 return;
70 76
71 /* Determine space needed for whole changeset */ 77 /* ... and pointer to end of current output buffer.
72 for( bucket = 0; bucket < OT_BUCKET_COUNT; ++bucket ) { 78 This works as a low watermark */
73 torrents_list = mutex_bucket_lock(bucket); 79 re = r + OT_SYNC_CHUNK_SIZE;
74 for( i=0; i<torrents_list->size; ++i ) {
75 ot_torrent *torrent = ((ot_torrent*)(torrents_list->data)) + i;
76 allocated += sizeof( ot_hash ) + sizeof(ot_peer) * torrent->peer_list->changeset.size + 13;
77 }
78 mutex_bucket_unlock(bucket);
79 }
80 80
81 /* add "d4:syncd" and "ee" */ 81 memmove( r, "d4:syncd", 8 ); r += 8;
82 allocated += 8 + 2;
83 82
84 if( !( r = *reply = mmap( NULL, allocated, PROT_READ | PROT_WRITE, MAP_ANON | MAP_PRIVATE, -1, 0 ) ) ) 83 /* For each bucket... */
85 return 0; 84 for( bucket=0; bucket<OT_BUCKET_COUNT; ++bucket ) {
85 /* Get exclusive access to that bucket */
86 ot_vector *torrents_list = mutex_bucket_lock( bucket );
87 size_t tor_offset;
88
89 /* For each torrent in this bucket.. */
90 for( tor_offset=0; tor_offset<torrents_list->size; ++tor_offset ) {
91 /* Address torrents members */
92 ot_peerlist *peer_list = ( ((ot_torrent*)(torrents_list->data))[tor_offset] ).peer_list;
93 ot_hash *hash =&( ((ot_torrent*)(torrents_list->data))[tor_offset] ).hash;
94 const size_t byte_count = sizeof(ot_peer) * peer_list->changeset.size;
95
96 /* If we reached our low watermark in buffer... */
97 if( re - r <= (ssize_t)(/* strlen( "20:" ) == */ 3 + sizeof( ot_hash ) + /* strlen_max( "%zd" ) == */ 12 + byte_count ) ) {
98
99 /* crop current output buffer to the amount really used */
100 iovec_fixlast( iovec_entries, iovector, OT_SYNC_CHUNK_SIZE - ( re - r ) );
101
102 /* And allocate a fresh output buffer at the end of our buffers list */
103 if( !( r = iovec_increase( iovec_entries, iovector, OT_SYNC_CHUNK_SIZE ) ) ) {
104
105 /* If this fails: free buffers */
106 iovec_free( iovec_entries, iovector );
107
108 /* Release lock on current bucket and return */
109 mutex_bucket_unlock( bucket );
110 return;
111 }
112
113 /* Adjust new end of output buffer */
114 re = r + OT_SYNC_CHUNK_SIZE;
115 }
86 116
87 memmove( r, "d4:syncd", 8 ); r += 8;
88 for( bucket = 0; bucket < OT_BUCKET_COUNT; ++bucket ) {
89 torrents_list = mutex_bucket_lock(bucket);
90 for( i=0; i<torrents_list->size; ++i ) {
91 ot_torrent *torrent = ((ot_torrent*)(torrents_list->data)) + i;
92 const size_t byte_count = sizeof(ot_peer) * torrent->peer_list->changeset.size;
93 *r++ = '2'; *r++ = '0'; *r++ = ':'; 117 *r++ = '2'; *r++ = '0'; *r++ = ':';
94 memmove( r, torrent->hash, sizeof( ot_hash ) ); r += sizeof( ot_hash ); 118 memmove( r, hash, sizeof( ot_hash ) ); r += sizeof( ot_hash );
95 r += sprintf( r, "%zd:", byte_count ); 119 r += sprintf( r, "%zd:", byte_count );
96 memmove( r, torrent->peer_list->changeset.data, byte_count ); r += byte_count; 120 memmove( r, peer_list->changeset.data, byte_count ); r += byte_count;
97 } 121 }
98 mutex_bucket_unlock(bucket); 122
123 /* All torrents done: release lock on currenct bucket */
124 mutex_bucket_unlock( bucket );
99 } 125 }
100 *r++ = 'e'; *r++ = 'e';
101 126
102 replysize = ( r - *reply ); 127 /* Close bencoded sync dictionary */
103 fix_mmapallocation( *reply, allocated, replysize ); 128 *r++='e'; *r++='e';
104 129
105 return replysize; 130 /* Release unused memory in current output buffer */
131 iovec_fixlast( iovec_entries, iovector, OT_SYNC_CHUNK_SIZE - ( re - r ) );
106} 132}
133
134/* This is the entry point into this worker thread
135 It grabs tasks from mutex_tasklist and delivers results back
136*/
137static void * sync_worker( void * args) {
138 int iovec_entries;
139 struct iovec *iovector;
140
141 args = args;
142
143 while( 1 ) {
144 ot_tasktype tasktype = TASK_SYNC_OUT;
145 ot_taskid taskid = mutex_workqueue_poptask( &tasktype );
146 sync_make( &iovec_entries, &iovector );
147 if( mutex_workqueue_pushresult( taskid, iovec_entries, iovector ) )
148 iovec_free( &iovec_entries, &iovector );
149 }
150 return NULL;
151}
152
153static pthread_t thread_id;
154void sync_init( ) {
155 pthread_create( &thread_id, NULL, sync_worker, NULL );
156}
157
158void sync_deinit( ) {
159 pthread_cancel( thread_id );
160}
161
162void sync_deliver( int64 socket ) {
163 mutex_workqueue_pushtask( socket, TASK_SYNC_OUT );
164}
165
107#endif 166#endif